Merge pull request #4093 from omnivore-app/fix/home-feed-job

fix/home feed job

packages/api/src/data_source.ts

@@ -24,3 +24,26 @@ export const appDataSource = new DataSource({
     idleTimeoutMillis: 10000, // 10 seconds
   },
 })
+
+if (env.pg.replication) {
+  appDataSource.setOptions({
+    replication: {
+      master: {
+        host: env.pg.host,
+        port: env.pg.port,
+        username: env.pg.userName,
+        password: env.pg.password,
+        database: env.pg.dbName,
+      },
+      slaves: [
+        {
+          host: env.pg.slave.host,
+          port: env.pg.slave.port,
+          username: env.pg.slave.userName,
+          password: env.pg.slave.password,
+          database: env.pg.slave.dbName,
+        },
+      ],
+    },
+  })
+}

packages/api/src/jobs/score_library_item.ts

@@ -1,5 +1,10 @@
-import { findLibraryItemById } from '../services/library_item'
+import { SubscriptionType } from '../entity/subscription'
+import {
+  findLibraryItemById,
+  updateLibraryItem,
+} from '../services/library_item'
 import { Feature, scoreClient } from '../services/score'
+import { findSubscriptionsByNames } from '../services/subscriptions'
 import { enqueueUpdateHomeJob } from '../utils/createTask'
 import { lanaugeToCode } from '../utils/helpers'
 import { logger } from '../utils/logger'
@@ -31,6 +36,8 @@ export const scoreLibraryItem = async (
       'author',
       'itemLanguage',
       'wordCount',
+      'subscription',
+      'publishedAt',
     ],
   })
   if (!libraryItem) {
@@ -38,6 +45,26 @@ export const scoreLibraryItem = async (
     return
   }
 
+  let subscription
+  if (libraryItem.subscription) {
+    const subscriptions = await findSubscriptionsByNames(userId, [
+      libraryItem.subscription,
+    ])
+
+    if (subscriptions.length) {
+      subscription = subscriptions[0]
+
+      if (subscription.type === SubscriptionType.Rss) {
+        logger.info('Skipping scoring for RSS subscription', {
+          userId,
+          libraryItemId,
+        })
+
+        return
+      }
+    }
+  }
+
   const itemFeatures = {
     [libraryItem.id]: {
       library_item_id: libraryItem.id,
@@ -53,7 +80,15 @@ export const scoreLibraryItem = async (
       language: lanaugeToCode(libraryItem.itemLanguage || 'English'),
       word_count: libraryItem.wordCount,
       published_at: libraryItem.publishedAt,
-      subscription: libraryItem.subscription,
+      subscription: subscription?.name,
+      inbox_folder: libraryItem.folder === 'inbox',
+      is_feed: subscription?.type === SubscriptionType.Rss,
+      is_newsletter: subscription?.type === SubscriptionType.Newsletter,
+      is_subscription: !!subscription,
+      item_word_count: libraryItem.wordCount,
+      subscription_auto_add_to_library: subscription?.autoAddToLibrary,
+      subscription_fetch_content: subscription?.fetchContent,
+      subscription_count: 0,
     } as Feature,
   }
 
@@ -69,15 +104,15 @@ export const scoreLibraryItem = async (
     throw new Error('Failed to score library item')
   }
 
-  // await updateLibraryItem(
-  //   libraryItem.id,
-  //   {
-  //     score,
-  //   },
-  //   userId,
-  //   undefined,
-  //   true
-  // )
+  await updateLibraryItem(
+    libraryItem.id,
+    {
+      score,
+    },
+    userId,
+    undefined,
+    true
+  )
   logger.info('Library item scored', data)
 
   try {

packages/api/src/services/score.ts

@@ -79,4 +79,4 @@ class ScoreClientImpl implements ScoreClient {
   }
 }
 
-export const scoreClient = new StubScoreClientImpl()
+export const scoreClient = new ScoreClientImpl()

packages/api/src/util.ts

@@ -19,6 +19,14 @@ export interface BackendEnv {
     pool: {
       max: number
     }
+    replication: boolean
+    slave: {
+      host: string
+      port: number
+      userName: string
+      password: string
+      dbName: string
+    }
   }
   server: {
     jwtSecret: string
@@ -179,6 +187,12 @@ const nullableEnvVars = [
   'NOTION_CLIENT_SECRET',
   'NOTION_AUTH_URL',
   'SCORE_API_URL',
+  'PG_REPLICATION',
+  'PG_SLAVE_HOST',
+  'PG_SLAVE_PORT',
+  'PG_SLAVE_USER',
+  'PG_SLAVE_PASSWORD',
+  'PG_SLAVE_DB',
 ] // Allow some vars to be null/empty
 
 const envParser =
@@ -218,6 +232,14 @@ export function getEnv(): BackendEnv {
     pool: {
       max: parseInt(parse('PG_POOL_MAX'), 10),
     },
+    replication: parse('PG_REPLICATION') === 'true',
+    slave: {
+      host: parse('PG_SLAVE_HOST'),
+      port: parseInt(parse('PG_SLAVE_PORT'), 10),
+      userName: parse('PG_SLAVE_USER'),
+      password: parse('PG_SLAVE_PASSWORD'),
+      dbName: parse('PG_SLAVE_DB'),
+    },
   }
   const server = {
     jwtSecret: parse('JWT_SECRET'),

packages/db/migrations/0183.do.alter_omnivore_admin_role.sql

@@ -0,0 +1,36 @@
+-- Type: DO
+-- Name: alter_omnivore_admin_role
+-- Description: Alter omnivore_admin role to prevent omnivore_admin to be inherited by app_user or omnivore_user
+
+BEGIN;
+
+DROP POLICY user_admin_policy ON omnivore.user;
+
+REVOKE ALL PRIVILEGES ON ALL TABLES IN SCHEMA omnivore from omnivore_admin;
+REVOKE ALL PRIVILEGES ON SCHEMA omnivore from omnivore_admin;
+
+DROP OWNED BY omnivore_admin;
+
+DROP ROLE omnivore_admin;
+
+CREATE ROLE omnivore_admin;
+
+GRANT USAGE ON SCHEMA omnivore TO omnivore_admin;
+
+ALTER ROLE omnivore_user NOINHERIT; -- This is to prevent omnivore_user from inheriting omnivore_admin role
+
+GRANT omnivore_admin TO omnivore_user; -- This is to allow app_user to set omnivore_admin role
+
+GRANT SELECT, INSERT, UPDATE, DELETE ON omnivore.user TO omnivore_admin;
+CREATE POLICY user_admin_policy on omnivore.user
+    FOR ALL
+    TO omnivore_admin
+    USING (true);
+
+GRANT SELECT, INSERT, UPDATE, DELETE ON omnivore.library_item TO omnivore_admin;
+CREATE POLICY library_item_admin_policy ON omnivore.library_item 
+    FOR ALL
+    TO omnivore_admin
+    USING (true);
+
+COMMIT;

packages/db/migrations/0183.undo.alter_omnivore_admin_role.sql

@@ -0,0 +1,31 @@
+-- Type: UNDO
+-- Name: alter_omnivore_admin_role
+-- Description: Alter omnivore_admin role to prevent omnivore_admin to be inherited by app_user or omnivore_user
+
+BEGIN;
+
+DROP POLICY library_item_admin_policy ON omnivore.library_item;
+REVOKE SELECT, INSERT, UPDATE, DELETE ON omnivore.library_item FROM omnivore_admin;
+
+DROP POLICY user_admin_policy ON omnivore.user;
+REVOKE SELECT, INSERT, UPDATE, DELETE ON omnivore.user FROM omnivore_admin;
+
+DROP OWNED BY omnivore_admin;
+
+DROP ROLE omnivore_admin;
+
+ALTER ROLE omnivore_user INHERIT;
+
+CREATE ROLE omnivore_admin;
+
+GRANT omnivore_admin TO app_user;
+
+GRANT ALL PRIVILEGES ON SCHEMA omnivore TO omnivore_admin;
+GRANT ALL PRIVILEGES ON ALL TABLES IN SCHEMA omnivore TO omnivore_admin;
+
+CREATE POLICY user_admin_policy on omnivore.user
+    FOR ALL
+    TO omnivore_admin
+    USING (true);
+
+COMMIT;