send notifications to the channels defined in digest config

packages/api/src/entity/user.ts

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

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

@@ -12,6 +12,8 @@ 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'
@@ -19,12 +21,15 @@ import {
   findLibraryItemsByIds,
   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 +82,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
@@ -546,14 +553,86 @@ const uploadSummary = async (
   console.timeEnd('uploadSummary')
 }
 
+const sendPushNotification = async (userId: string, digest: Digest) => {
+  const notification = {
+    title: digest.title ?? 'Omnivore digest',
+    body: 'Your digest is ready to listen',
+  }
+
+  await sendPushNotifications(userId, notification, 'reminder')
+}
+
+const sendEmail = async (user: User, digest: Digest) => {
+  const title = digest.title ?? 'Omnivore digest'
+  const html = `
+    <h>${title}</p>
+
+    <h1>Transcript</h1>
+    <p>${digest.content ?? 'Transcript not available'}</p>
+  `
+
+  await enqueueSendEmail({
+    to: user.email,
+    from: env.sender.message,
+    subject: title,
+    html,
+  })
+}
+
+const sendNotifications = async (
+  user: User,
+  channels: Channel[],
+  digest: Digest
+) => {
+  await Promise.all(
+    channels.map(async (channel) => {
+      switch (channel) {
+        case 'push':
+          return sendPushNotification(user.id, digest)
+        case 'email':
+          return sendEmail(user, digest)
+        default:
+      }
+    })
+  )
+}
+
 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,
+  }
+
   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 +641,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)
@@ -591,7 +666,7 @@ export const createDigest = async (jobData: CreateDigestData) => {
       secondaryVoice: jobData.voices?.[1],
     })
     const title = generateTitle(summaries)
-    const digest: Digest = {
+    digest = {
       id: digestId,
       title,
       content: generateContent(summaries),
@@ -630,19 +705,10 @@ export const createDigest = async (jobData: CreateDigestData) => {
       jobState: TaskState.Failed,
     })
   } finally {
+    // default to push notification
+    const channels = config?.channels ?? ['push']
     // 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, channels, digest)
 
     console.timeEnd('createDigestJob')
   }

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