Merge pull request #3893 from omnivore-app/feature/digest-email

feature/digest email

packages/api/src/entity/user.ts

@@ -73,5 +73,5 @@ export class User {
     () => UserPersonalization,
     (userPersonalization) => userPersonalization.user
   )
-  userPersonalization!: UserPersonalization
+  userPersonalization?: UserPersonalization
 }

packages/api/src/entity/user_personalization.ts

@@ -56,4 +56,7 @@ export class UserPersonalization {
 
   @Column('json')
   fields?: any | null
+
+  @Column('jsonb')
+  digestConfig?: any | null
 }

packages/api/src/generated/graphql.ts

@@ -3054,6 +3054,7 @@ export enum SetUserPersonalizationErrorCode {
 }
 
 export type SetUserPersonalizationInput = {
+  digestConfig?: InputMaybe<Scalars['JSON']>;
   fields?: InputMaybe<Scalars['JSON']>;
   fontFamily?: InputMaybe<Scalars['String']>;
   fontSize?: InputMaybe<Scalars['Int']>;
@@ -3767,6 +3768,7 @@ export enum UserErrorCode {
 
 export type UserPersonalization = {
   __typename?: 'UserPersonalization';
+  digestConfig?: Maybe<Scalars['JSON']>;
   fields?: Maybe<Scalars['JSON']>;
   fontFamily?: Maybe<Scalars['String']>;
   fontSize?: Maybe<Scalars['Int']>;
@@ -7230,6 +7232,7 @@ export type UserErrorResolvers<ContextType = ResolverContext, ParentType extends
 };
 
 export type UserPersonalizationResolvers<ContextType = ResolverContext, ParentType extends ResolversParentTypes['UserPersonalization'] = ResolversParentTypes['UserPersonalization']> = {
+  digestConfig?: Resolver<Maybe<ResolversTypes['JSON']>, ParentType, ContextType>;
   fields?: Resolver<Maybe<ResolversTypes['JSON']>, ParentType, ContextType>;
   fontFamily?: Resolver<Maybe<ResolversTypes['String']>, ParentType, ContextType>;
   fontSize?: Resolver<Maybe<ResolversTypes['Int']>, ParentType, ContextType>;

packages/api/src/generated/schema.graphql

@@ -2385,6 +2385,7 @@ enum SetUserPersonalizationErrorCode {
 }
 
 input SetUserPersonalizationInput {
+  digestConfig: JSON
   fields: JSON
   fontFamily: String
   fontSize: Int
@@ -3044,6 +3045,7 @@ enum UserErrorCode {
 }
 
 type UserPersonalization {
+  digestConfig: JSON
   fields: JSON
   fontFamily: String
   fontSize: Int

packages/api/src/jobs/ai/create_digest.ts

@@ -8,23 +8,30 @@ import {
   SSMLOptions,
 } from '@omnivore/text-to-speech-handler'
 import axios from 'axios'
+import { truncate } from 'lodash'
 import showdown from 'showdown'
 import { v4 as uuid } from 'uuid'
 import yaml from 'yaml'
 import { LibraryItem } from '../../entity/library_item'
+import { User } from '../../entity/user'
+import { env } from '../../env'
 import { TaskState } from '../../generated/graphql'
 import { redisDataSource } from '../../redis_data_source'
 import { Digest, writeDigest } from '../../services/digest'
 import {
   findLibraryItemsByIds,
+  getItemUrl,
   searchLibraryItems,
 } from '../../services/library_item'
-import { findDeviceTokensByUserId } from '../../services/user_device_tokens'
+import {
+  findUserAndPersonalization,
+  sendPushNotifications,
+} from '../../services/user'
+import { enqueueSendEmail } from '../../utils/createTask'
 import { wordsCount } from '../../utils/helpers'
 import { logger } from '../../utils/logger'
 import { htmlToMarkdown } from '../../utils/parser'
-import { sendMulticastPushNotifications } from '../../utils/sendNotification'
-import { generateUploadFilePathName, uploadToBucket } from '../../utils/uploads'
+import { uploadToBucket } from '../../utils/uploads'
 
 export type CreateDigestJobSchedule = 'daily' | 'weekly'
 
@@ -77,6 +84,8 @@ interface RankedTitle {
   title: string
 }
 
+type Channel = 'push' | 'email'
+
 export const CREATE_DIGEST_JOB = 'create-digest'
 export const CRON_PATTERNS = {
   // every day at 10:30 UTC
@@ -524,7 +533,7 @@ const uploadSummary = async (
   console.time('uploadSummary')
   logger.info('uploading summaries to gcs')
 
-  const filename = `digest/${userId}/${digest.id}/summaries.json`
+  const filename = `digest/${userId}/${digest.id}.json`
   await uploadToBucket(
     filename,
     Buffer.from(
@@ -546,14 +555,116 @@ const uploadSummary = async (
   console.timeEnd('uploadSummary')
 }
 
+const sendPushNotification = async (userId: string, digest: Digest) => {
+  const notification = {
+    title: 'Omnivore Digest',
+    body: truncate(digest.title, { length: 100 }),
+  }
+
+  await sendPushNotifications(userId, notification, 'reminder')
+}
+
+const sendEmail = async (
+  user: User,
+  digest: Digest,
+  summaries: RankedItem[]
+) => {
+  const createdAt = digest.createdAt ?? new Date()
+
+  const prefix = 'Omnivore Digest'
+  const title = `${prefix} ${createdAt.toLocaleDateString()}`
+  const subTitle = truncate(digest.title, { length: 200 }).slice(
+    prefix.length + 1
+  )
+
+  const chapters = digest.chapters ?? []
+
+  const html = `
+    <div style="text-align: justify;">
+      <h2>${title}</h1>
+      <h2>${subTitle}</h2>
+
+        ${chapters
+          .map(
+            (chapter, index) => `
+              <div>
+                <a href="${chapter.url}"><h3>${chapter.title} (${chapter.wordCount} words)</h3></a>
+                <div>
+                  ${summaries[index].summary}
+                </div>
+              </div>`
+          )
+          .join('')}
+    </div>`
+
+  await enqueueSendEmail({
+    to: user.email,
+    from: env.sender.message,
+    subject: subTitle,
+    html,
+  })
+}
+
+const sendNotifications = async (
+  user: User,
+  digest: Digest,
+  summaries: RankedItem[],
+  channels: Channel[] = ['push'] // default to push notification
+) => {
+  const deduplicateChannels = [...new Set(channels)]
+
+  await Promise.all(
+    deduplicateChannels.map(async (channel) => {
+      switch (channel) {
+        case 'push':
+          return sendPushNotification(user.id, digest)
+        case 'email':
+          return sendEmail(user, digest, summaries)
+        default:
+          logger.error('Unknown channel', { channel })
+          return
+      }
+    })
+  )
+}
+
 export const createDigest = async (jobData: CreateDigestData) => {
   console.time('createDigestJob')
 
   // generate a unique id for the digest if not provided for scheduled jobs
   const digestId = jobData.id ?? uuid()
+
+  const user = await findUserAndPersonalization(jobData.userId)
+  if (!user) {
+    logger.error('User not found', { userId: jobData.userId })
+    return writeDigest(jobData.userId, {
+      id: digestId,
+      jobState: TaskState.Failed,
+    })
+  }
+
+  const personalization = user.userPersonalization
+  if (!personalization) {
+    logger.info('User personalization not found')
+  }
+
+  const config = personalization
+    ? (personalization.digestConfig as {
+        model?: string
+        channels?: Channel[]
+      })
+    : undefined
+
+  // default digest
+  let digest: Digest = {
+    id: digestId,
+    jobState: TaskState.Succeeded,
+  }
+  let filteredSummaries: RankedItem[] = []
+
   try {
     digestDefinition = await fetchDigestDefinition()
-    const model = selectModel(digestDefinition.model)
+    const model = selectModel(config?.model || digestDefinition.model)
     logger.info(`model: ${model}`)
 
     const candidates = await getCandidatesList(
@@ -562,11 +673,7 @@ export const createDigest = async (jobData: CreateDigestData) => {
     )
     if (candidates.length === 0) {
       logger.info('No candidates found')
-      return writeDigest(jobData.userId, {
-        id: digestId,
-        jobState: TaskState.Succeeded,
-        title: 'No articles found',
-      })
+      return writeDigest(jobData.userId, digest)
     }
 
     // const userProfile = await findOrCreateUserProfile(jobData.userId)
@@ -583,31 +690,31 @@ export const createDigest = async (jobData: CreateDigestData) => {
     const summaries = await summarizeItems(model, selections)
     console.timeEnd('summarizeItems')
 
-    const filteredSummaries = filterSummaries(summaries)
+    filteredSummaries = filterSummaries(summaries)
 
     const speechFiles = generateSpeechFiles(filteredSummaries, {
       ...jobData,
       primaryVoice: jobData.voices?.[0],
       secondaryVoice: jobData.voices?.[1],
     })
-    const title = generateTitle(summaries)
-    const digest: Digest = {
+    const title = generateTitle(filteredSummaries)
+    digest = {
       id: digestId,
       title,
-      content: generateContent(summaries),
+      content: generateContent(filteredSummaries),
       jobState: TaskState.Succeeded,
       speechFiles,
       chapters: filteredSummaries.map((item, index) => ({
         title: item.libraryItem.title,
         id: item.libraryItem.id,
-        url: item.libraryItem.originalUrl,
+        url: getItemUrl(item.libraryItem.id),
         thumbnail: item.libraryItem.thumbnail ?? undefined,
         wordCount: speechFiles[index].wordCount,
       })),
       createdAt: new Date(),
       description: '',
-      // description: generateDescription(summaries, rankedTopics),
-      byline: generateByline(summaries),
+      // description: generateDescription(filteredSummaries, rankedTopics),
+      byline: generateByline(filteredSummaries),
       urlsToAudio: [],
       model,
     }
@@ -616,7 +723,7 @@ export const createDigest = async (jobData: CreateDigestData) => {
       // write the digest to redis
       writeDigest(jobData.userId, digest),
       // upload the summaries to GCS
-      uploadSummary(jobData.userId, digest, summaries).catch((error) =>
+      uploadSummary(jobData.userId, digest, filteredSummaries).catch((error) =>
         logger.error('uploadSummary error', error)
       ),
     ])
@@ -628,21 +735,11 @@ export const createDigest = async (jobData: CreateDigestData) => {
     await writeDigest(jobData.userId, {
       id: digestId,
       jobState: TaskState.Failed,
+      title: 'Failed to create digest',
     })
   } 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')
-    }
+    await sendNotifications(user, digest, filteredSummaries, config?.channels)
 
     console.timeEnd('createDigestJob')
   }

packages/api/src/schema.ts

@@ -1080,6 +1080,7 @@ const schema = gql`
     speechRate: String
     speechVolume: String
     fields: JSON
+    digestConfig: JSON
   }
 
   # Query: UserPersonalization
@@ -1122,6 +1123,7 @@ const schema = gql`
     speechRate: String
     speechVolume: String
     fields: JSON
+    digestConfig: JSON
   }
 
   # Type: ArticleSavingRequest

packages/api/src/services/user.ts

@@ -150,3 +150,17 @@ export const sendPushNotifications = async (
 
   return sendMulticastPushNotifications(userId, message, notificationType)
 }
+
+export const findUserAndPersonalization = async (id: string) => {
+  return authTrx(
+    (t) =>
+      t.getRepository(User).findOne({
+        where: { id },
+        relations: {
+          userPersonalization: true,
+        },
+      }),
+    undefined,
+    id
+  )
+}

packages/api/src/services/user_personalization.ts

@@ -2,22 +2,22 @@ import { DeepPartial } from 'typeorm'
 import { UserPersonalization } from '../entity/user_personalization'
 import { authTrx } from '../repository'
 
-export const findUserPersonalization = async (id: string, userId: string) => {
+export const findUserPersonalization = async (userId: string) => {
   return authTrx(
     (t) =>
       t.getRepository(UserPersonalization).findOneBy({
-        id,
+        user: { id: userId },
       }),
     undefined,
     userId
   )
 }
 
-export const deleteUserPersonalization = async (id: string, userId: string) => {
+export const deleteUserPersonalization = async (userId: string) => {
   return authTrx(
     (t) =>
       t.getRepository(UserPersonalization).delete({
-        id,
+        user: { id: userId },
       }),
     undefined,
     userId

packages/api/test/resolvers/user_personalization.test.ts

@@ -22,7 +22,7 @@ describe('User Personalization API', () => {
       .post('/local/debug/fake-user-login')
       .send({ fakeEmail: user.email })
 
-    authToken = res.body.authToken
+    authToken = res.body.authToken as string
   })
 
   after(async () => {
@@ -61,40 +61,32 @@ describe('User Personalization API', () => {
           res.body.data.setUserPersonalization.updatedUserPersonalization.fields
         ).to.eql(fields)
 
-        const userPersonalization = await findUserPersonalization(
-          res.body.data.setUserPersonalization.updatedUserPersonalization.id,
-          user.id
-        )
+        const userPersonalization = await findUserPersonalization(user.id)
         expect(userPersonalization).to.not.be.null
 
         // clean up
-        await deleteUserPersonalization(
-          res.body.data.setUserPersonalization.updatedUserPersonalization.id,
-          user.id
-        )
+        await deleteUserPersonalization(user.id)
       })
     })
 
     context('when user personalization exists', () => {
-      let existingUserPersonalization: UserPersonalization
-
       before(async () => {
-        existingUserPersonalization = await saveUserPersonalization(user.id, {
+        await saveUserPersonalization(user.id, {
           user: { id: user.id },
-          fields: {
-            testField: 'testValue',
+          digestConfig: {
+            channels: ['email'],
           },
         })
       })
 
       after(async () => {
         // clean up
-        await deleteUserPersonalization(existingUserPersonalization.id, user.id)
+        await deleteUserPersonalization(user.id)
       })
 
       it('updates the user personalization', async () => {
         const newFields = {
-          testField: 'testValue1',
+          channels: ['push', 'email'],
         }
 
         const res = await graphqlRequest(query, authToken, {
@@ -106,7 +98,6 @@ describe('User Personalization API', () => {
         ).to.eql(newFields)
 
         const updatedUserPersonalization = await findUserPersonalization(
-          existingUserPersonalization.id,
           user.id
         )
         expect(updatedUserPersonalization?.fields).to.eql(newFields)
@@ -128,7 +119,7 @@ describe('User Personalization API', () => {
 
     after(async () => {
       // clean up
-      await deleteUserPersonalization(existingUserPersonalization.id, user.id)
+      await deleteUserPersonalization(user.id)
     })
 
     const query = `
@@ -150,10 +141,9 @@ describe('User Personalization API', () => {
     it('returns the user personalization', async () => {
       const res = await graphqlRequest(query, authToken).expect(200)
 
-      expect(res.body.data.getUserPersonalization.userPersonalization).to.eql({
-        id: existingUserPersonalization.id,
-        fields: existingUserPersonalization.fields,
-      })
+      expect(
+        res.body.data.getUserPersonalization.userPersonalization.fields
+      ).to.eql(existingUserPersonalization.fields)
     })
   })
 })

packages/db/migrations/0174.do.add_digest_config_to_user_personalization.sql

@@ -0,0 +1,9 @@
+-- Type: DO
+-- Name: add_digest_config_to_user_personalization
+-- Description: Add digest_config json column to the user_personalization table
+
+BEGIN;
+
+ALTER TABLE omnivore.user_personalization ADD COLUMN digest_config jsonb;
+
+COMMIT;

packages/db/migrations/0174.undo.add_digest_config_to_user_personalization.sql

@@ -0,0 +1,9 @@
+-- Type: UNDO
+-- Name: add_digest_config_to_user_personalization
+-- Description: Add digest_config json column to the user_personalization table
+
+BEGIN;
+
+ALTER TABLE omnivore.user_personalization DROP COLUMN digest_config;
+
+COMMIT;