implement updateJustReadFeed job

packages/api/src/entity/public_item.ts

@@ -2,22 +2,20 @@ import {
   Column,
   CreateDateColumn,
   Entity,
-  OneToOne,
   PrimaryGeneratedColumn,
   UpdateDateColumn,
 } from 'typeorm'
-import { PublicItemInteraction } from './public_item_interaction'
 
 @Entity()
 export class PublicItem {
   @PrimaryGeneratedColumn('uuid')
   id!: string
 
-  @OneToOne(() => PublicItemInteraction)
-  interaction?: PublicItemInteraction
+  @Column('text')
+  source_name!: string
 
-  @Column('uuid')
-  sourceId!: string
+  @Column('text')
+  source_icon!: string
 
   @Column('text')
   type!: string
@@ -31,23 +29,23 @@ export class PublicItem {
   @Column('boolean')
   approved!: boolean
 
-  @Column('text', { nullable: true })
-  thumbnail?: string | null
+  @Column('text')
+  thumbnail?: string
 
-  @Column('text', { nullable: true })
-  previewContent?: string | null
+  @Column('text')
+  previewContent?: string
 
-  @Column('text', { nullable: true })
-  languageCode?: string | null
+  @Column('text')
+  languageCode?: string
 
-  @Column('text', { nullable: true })
-  author?: string | null
+  @Column('text')
+  author?: string
 
-  @Column('text', { nullable: true })
-  dir?: string | null
+  @Column('text')
+  dir?: string
 
-  @Column('timestamptz', { nullable: true })
-  publishedAt?: Date | null
+  @Column('timestamptz')
+  publishedAt?: Date
 
   @CreateDateColumn()
   createdAt!: Date

packages/api/src/entity/public_item_interaction.ts

@@ -3,7 +3,6 @@ import {
   Entity,
   JoinColumn,
   ManyToOne,
-  OneToOne,
   PrimaryGeneratedColumn,
 } from 'typeorm'
 import { PublicItem } from './public_item'
@@ -14,16 +13,19 @@ export class PublicItemInteraction {
   @PrimaryGeneratedColumn('uuid')
   id!: string
 
-  @OneToOne(() => PublicItem, { onDelete: 'CASCADE' })
+  @ManyToOne(() => PublicItem, { onDelete: 'CASCADE' })
   @JoinColumn({ name: 'public_item_id' })
   publicItem!: PublicItem
 
+  @Column('uuid')
+  publicItemId!: string
+
   @ManyToOne(() => User, { onDelete: 'CASCADE' })
   @JoinColumn({ name: 'user_id' })
   user!: User
 
   @Column('timestamptz')
-  seenAt?: Date
+  seenAt!: Date
 
   @Column('timestamptz')
   savedAt?: Date

packages/api/src/jobs/update_just_read_feed.ts

@@ -1,3 +1,6 @@
+import { LibraryItem } from '../entity/library_item'
+import { PublicItem } from '../entity/public_item'
+import { redisDataSource } from '../redis_data_source'
 import { searchLibraryItems } from '../services/library_item'
 import { findUnseenPublicItems } from '../services/public_item'
 import { logger } from '../utils/logger'
@@ -6,9 +9,48 @@ interface JustReadFeedUpdateData {
   userId: string
 }
 
-const selectCandidates = async (userId: string) => {
+interface Candidate {
+  id: string
+  title: string
+  url: string
+  thumbnail?: string
+  previewContent?: string
+  languageCode?: string
+  author?: string
+  dir?: string
+  publishedAt?: Date
+  subscription?: string
+}
+
+const libraryItemToCandidate = (item: LibraryItem): Candidate => ({
+  id: item.id,
+  title: item.title,
+  url: item.originalUrl,
+  thumbnail: item.thumbnail || undefined,
+  previewContent: item.description || undefined,
+  languageCode: item.itemLanguage || undefined, // TODO: map to language code
+  author: item.author || undefined,
+  dir: item.directionality || undefined,
+  publishedAt: item.publishedAt || undefined,
+  subscription: item.subscription || undefined,
+})
+
+const publicItemToCandidate = (item: PublicItem): Candidate => ({
+  id: item.id,
+  title: item.title,
+  url: item.url,
+  thumbnail: item.thumbnail,
+  previewContent: item.previewContent,
+  languageCode: item.languageCode,
+  author: item.author,
+  dir: item.dir,
+  publishedAt: item.publishedAt,
+  subscription: item.source_name,
+})
+
+const selectCandidates = async (userId: string): Promise<Array<Candidate>> => {
   // get last 100 library items saved and not seen by user
-  const privateCandidates = await searchLibraryItems(
+  const libraryItems = await searchLibraryItems(
     {
       size: 100,
       includeContent: false,
@@ -17,13 +59,65 @@ const selectCandidates = async (userId: string) => {
     userId
   )
 
+  // map library items to candidates
+  const privateCandidates: Array<Candidate> = libraryItems.map(
+    libraryItemToCandidate
+  )
+
   // get candidates from public inventory
-  const publicCandidates = await findUnseenPublicItems(userId, {
+  const publicItems = await findUnseenPublicItems(userId, {
     limit: 100,
   })
 
-  // TODO: mix candidates
-  return privateCandidates.concat(publicCandidates)
+  // map public items to candidates
+  const publicCandidates: Array<Candidate> = publicItems.map(
+    publicItemToCandidate
+  )
+
+  // combine candidates
+  return [...privateCandidates, ...publicCandidates]
+}
+
+const rankCandidates = async (
+  candidates: Array<Candidate>
+): Promise<Array<Candidate>> => {
+  if (candidates.length <= 10) {
+    return candidates
+  }
+
+  // TODO: rank candidates
+  const API_URL = 'https://rank.omnivore.app'
+
+  const response = await fetch(API_URL, {
+    method: 'POST',
+    headers: {
+      'Content-Type': 'application/json',
+    },
+    body: JSON.stringify({ candidates }),
+  })
+
+  if (!response.ok) {
+    throw new Error(`Failed to rank candidates: ${response.statusText}`)
+  }
+
+  return response.json() as Promise<Array<Candidate>>
+}
+
+const prependCandidatesToFeed = async (
+  candidates: Array<Candidate>,
+  userId: string
+) => {
+  const redisKey = `just-read-feed:${userId}`
+  const redisClient = redisDataSource.redisClient
+  if (!redisClient) {
+    throw new Error('Redis client not available')
+  }
+
+  const pipeline = redisClient.pipeline()
+  candidates.forEach((candidate) =>
+    pipeline.lpush(redisKey, JSON.stringify(candidate))
+  )
+  await pipeline.exec()
 }
 
 const updateJustReadFeed = async (data: JustReadFeedUpdateData) => {
@@ -34,7 +128,8 @@ const updateJustReadFeed = async (data: JustReadFeedUpdateData) => {
   logger.info(`Found ${candidates.length} candidates`)
 
   // TODO: integrity check on candidates?
-  // TODO: rank candidates
 
-  // TODO: prepend candidates to feed in redis
+  const rankedCandidates = await rankCandidates(candidates)
+
+  await prependCandidatesToFeed(rankedCandidates, userId)
 }

packages/api/src/services/public_item.ts

@@ -1,4 +1,3 @@
-import { IsNull } from 'typeorm'
 import { PublicItem } from '../entity/public_item'
 import { getRepository } from '../repository'
 
@@ -8,21 +7,17 @@ export const findUnseenPublicItems = async (
     limit?: number
     offset?: number
   }
-) =>
-  getRepository(PublicItem).find({
-    where: {
-      interaction: IsNull(),
-      interaction: {
-        user: {
-          id: userId,
-        },
-        seenAt: IsNull(),
-      },
-      approved: true,
-    },
-    order: {
-      createdAt: 'DESC',
-    },
-    take: options.limit,
-    skip: options.offset,
-  })
+): Promise<Array<PublicItem>> =>
+  getRepository(PublicItem)
+    .createQueryBuilder('public_item')
+    .leftJoin(
+      'omnivore.public_item_interactions',
+      'interaction',
+      'interaction.public_item_id = public_item.id'
+    )
+    .where('interaction.user_id = :userId', { userId })
+    .andWhere('interaction.seen_at IS NULL')
+    .orderBy('public_item.created_at', 'DESC')
+    .limit(options.limit)
+    .offset(options.offset)
+    .getMany()

packages/db/migrations/0177.do.public_item.sql

@@ -21,7 +21,8 @@ CREATE TRIGGER update_public_item_source_modtime BEFORE UPDATE ON omnivore.publi
 
 CREATE TABLE omnivore.public_item (
 	id uuid PRIMARY KEY DEFAULT uuid_generate_v1mc(),
-    source_id uuid NOT NULL, -- user_id or public_item_source_id
+    source_name TEXT NOT NULL,
+    source_icon TEXT,
     type TEXT NOT NULL, -- public feeds, newsletters, or user recommended
     title TEXT NOT NULL,
     url TEXT NOT NULL,
@@ -60,7 +61,7 @@ CREATE TABLE omnivore.public_item_interactions (
     saved_at TIMESTAMPTZ,
     liked_at TIMESTAMPTZ,
     broadcasted_at TIMESTAMPTZ,
-    seen_at TIMESTAMPTZ,
+    seen_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
     created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
     updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
 );