summaries

packages/api/package.json

@@ -77,6 +77,7 @@
     "ioredis": "^5.3.2",
     "jsonwebtoken": "^8.5.1",
     "jwks-rsa": "^2.0.3",
+    "langchain": "^0.1.21",
     "linkedom": "^0.14.9",
     "lodash": "^4.17.21",
     "luxon": "^3.2.1",

packages/api/src/pubsub.ts

@@ -5,12 +5,18 @@ import { env } from './env'
 import { ReportType } from './generated/graphql'
 import { Merge } from './util'
 import {
+  enqueueAISummarizeJob,
   enqueueExportItem,
   enqueueTriggerRuleJob,
   enqueueWebhookJob,
 } from './utils/createTask'
 import { deepDelete } from './utils/helpers'
 import { buildLogger } from './utils/logger'
+import {
+  FeatureName,
+  findFeatureByName,
+  getFeatureName,
+} from './services/features'
 
 const logger = buildLogger('pubsub')
 
@@ -82,6 +88,13 @@ export const createPubSubClient = (): PubsubClient => {
         data,
       })
 
+      if (await findFeatureByName(FeatureName.AISummaries, userId)) {
+        await enqueueAISummarizeJob({
+          userId,
+          libraryItemId,
+        })
+      }
+
       return publish(
         'entityCreated',
         Buffer.from(JSON.stringify({ type, userId, ...cleanData }))

packages/api/src/queue-processor.ts

@@ -43,6 +43,7 @@ import { redisDataSource } from './redis_data_source'
 import { CACHED_READING_POSITION_PREFIX } from './services/cached_reading_position'
 import { getJobPriority } from './utils/createTask'
 import { logger } from './utils/logger'
+import { AI_SUMMARIZE_JOB_NAME, aiSummarize } from './jobs/ai-summarize'
 
 export const QUEUE_NAME = 'omnivore-backend-queue'
 export const JOB_VERSION = 'v001'
@@ -113,6 +114,8 @@ export const createWorker = (connection: ConnectionOptions) =>
           return callWebhook(job.data)
         case EXPORT_ITEM_JOB_NAME:
           return exportItem(job.data)
+        case AI_SUMMARIZE_JOB_NAME:
+          return aiSummarize(job.data)
         case EXPORT_ALL_ITEMS_JOB_NAME:
           return exportAllItems(job.data)
       }

packages/api/src/services/features.ts

@@ -7,6 +7,7 @@ import { getRepository } from '../repository'
 import { logger } from '../utils/logger'
 
 export enum FeatureName {
+  AISummaries = 'ai-summaries',
   UltraRealisticVoice = 'ultra-realistic-voice',
 }
 
@@ -21,7 +22,6 @@ export const optInFeature = async (
   if (name === FeatureName.UltraRealisticVoice) {
     return optInUltraRealisticVoice(uid)
   }
-
   return undefined
 }
 

packages/api/src/utils/createTask.ts

@@ -44,6 +44,7 @@ import { CreateTaskError } from './errors'
 import { stringToHash } from './helpers'
 import { logger } from './logger'
 import View = google.cloud.tasks.v2.Task.View
+import { AISummarizeJobData, AI_SUMMARIZE_JOB_NAME } from '../jobs/ai-summarize'
 
 // Instantiates a client.
 const client = new CloudTasksClient()
@@ -66,6 +67,7 @@ export const getJobPriority = (jobName: string): number => {
     case TRIGGER_RULE_JOB_NAME:
     case CALL_WEBHOOK_JOB_NAME:
     case EXPORT_ITEM_JOB_NAME:
+    case AI_SUMMARIZE_JOB_NAME:
       return 5
     case BULK_ACTION_JOB_NAME:
     case `${REFRESH_FEED_JOB_NAME}_high`:
@@ -694,6 +696,18 @@ export const enqueueWebhookJob = async (data: CallWebhookJobData) => {
   })
 }
 
+export const enqueueAISummarizeJob = async (data: AISummarizeJobData) => {
+  const queue = await getBackendQueue()
+  if (!queue) {
+    return undefined
+  }
+
+  return queue.add(AI_SUMMARIZE_JOB_NAME, data, {
+    priority: getJobPriority(AI_SUMMARIZE_JOB_NAME),
+    attempts: 3,
+  })
+}
+
 export const bulkEnqueueUpdateLabels = async (data: UpdateLabelsData[]) => {
   const queue = await getBackendQueue()
   if (!queue) {