From 755f483ba93c30ca2e7863ff069afbad730aa911 Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Tue, 11 Jun 2024 18:51:25 +0800
Subject: [PATCH 1/9] create a function to batch delete items in trash

---
 packages/api/src/jobs/prune_library_item.ts   |  0
 packages/api/src/services/library_item.ts     | 24 ++++++++++++++
 .../0181.do.batch_delete_trash_items.sql      | 32 +++++++++++++++++++
 .../0181.undo.batch_delete_trash_items.sql    |  9 ++++++
 4 files changed, 65 insertions(+)
 create mode 100644 packages/api/src/jobs/prune_library_item.ts
 create mode 100755 packages/db/migrations/0181.do.batch_delete_trash_items.sql
 create mode 100755 packages/db/migrations/0181.undo.batch_delete_trash_items.sql

diff --git a/packages/api/src/jobs/prune_library_item.ts b/packages/api/src/jobs/prune_library_item.ts
new file mode 100644
index 000000000..e69de29bb
diff --git a/packages/api/src/services/library_item.ts b/packages/api/src/services/library_item.ts
index fb5a5c2cf..8cc8914ec 100644
--- a/packages/api/src/services/library_item.ts
+++ b/packages/api/src/services/library_item.ts
@@ -1392,6 +1392,30 @@ export const batchDelete = async (criteria: FindOptionsWhere<LibraryItem>) => {
   return authTrx(async (t) => t.query(sql))
 }
 
+export const batchDeleteAllTrash = async () => {
+  const sql = `
+    DO $$
+    DECLARE
+        user_record RECORD;
+        user_cursor CURSOR FOR SELECT id FROM omnivore.user WHERE status = 'ACTIVE';  -- Adjust the condition as needed
+    BEGIN
+        OPEN user_cursor;
+
+        LOOP
+            FETCH NEXT FROM user_cursor INTO user_record;
+            EXIT WHEN NOT FOUND;
+
+            DELETE FROM omnivore.library_item WHERE user_id = user_record.id AND state = 'DELETED' AND deleted_at < '2023-01-01';
+
+            RETURN NEXT;
+        END LOOP;
+
+        CLOSE user_cursor;
+    END $$;`
+
+  return authTrx(async (t) => t.query(sql))
+}
+
 export const findLibraryItemIdsByLabelId = async (
   labelId: string,
   userId: string
diff --git a/packages/db/migrations/0181.do.batch_delete_trash_items.sql b/packages/db/migrations/0181.do.batch_delete_trash_items.sql
new file mode 100755
index 000000000..07b38d538
--- /dev/null
+++ b/packages/db/migrations/0181.do.batch_delete_trash_items.sql
@@ -0,0 +1,32 @@
+-- Type: DO
+-- Name: batch_delete_trash_items
+-- Description: Create a function to batch delete library items in trash
+
+BEGIN;
+
+CREATE OR REPLACE FUNCTION batch_delete_trash_items()
+RETURNS VOID AS $$
+DECLARE
+    user_record RECORD;
+    user_cursor CURSOR FOR
+    SELECT
+        id
+    FROM
+        omnivore.user
+    WHERE
+        status = 'ACTIVE';
+BEGIN
+    FOR user_record IN user_cursor LOOP
+        -- For Row Level Security
+        PERFORM omnivore.set_claims(user_record.id, 'omnivore_user');
+
+        DELETE FROM omnivore.library_item
+        WHERE
+            user_id = user_record.id
+            AND state = 'DELETED'
+            AND deleted_at < NOW() - INTERVAL '14 days';
+    END LOOP;
+END;
+$$ LANGUAGE plpgsql;
+
+COMMIT;
diff --git a/packages/db/migrations/0181.undo.batch_delete_trash_items.sql b/packages/db/migrations/0181.undo.batch_delete_trash_items.sql
new file mode 100755
index 000000000..836f8e533
--- /dev/null
+++ b/packages/db/migrations/0181.undo.batch_delete_trash_items.sql
@@ -0,0 +1,9 @@
+-- Type: UNDO
+-- Name: batch_delete_trash_items
+-- Description: Create a function to batch delete library items in trash
+
+BEGIN;
+
+DROP FUNCTION IF EXISTS batch_delete_trash_items();
+
+COMMIT;

From fc5b152c439204b59b65d0b2df37ea4ecf854378 Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Wed, 12 Jun 2024 16:50:59 +0800
Subject: [PATCH 2/9] update job name

---
 packages/api/src/jobs/score_library_item.ts | 2 +-
 packages/api/src/jobs/update_home.ts        | 2 +-
 packages/api/src/jobs/upload_content.ts     | 2 +-
 3 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/packages/api/src/jobs/score_library_item.ts b/packages/api/src/jobs/score_library_item.ts
index 0dc3152dc..702bba53c 100644
--- a/packages/api/src/jobs/score_library_item.ts
+++ b/packages/api/src/jobs/score_library_item.ts
@@ -7,7 +7,7 @@ import { enqueueUpdateHomeJob } from '../utils/createTask'
 import { lanaugeToCode } from '../utils/helpers'
 import { logger } from '../utils/logger'
 
-export const SCORE_LIBRARY_ITEM_JOB = 'SCORE_LIBRARY_ITEM_JOB'
+export const SCORE_LIBRARY_ITEM_JOB = 'score-library-item'
 
 export interface ScoreLibraryItemJobData {
   userId: string
diff --git a/packages/api/src/jobs/update_home.ts b/packages/api/src/jobs/update_home.ts
index c80e18656..72e39f3d5 100644
--- a/packages/api/src/jobs/update_home.ts
+++ b/packages/api/src/jobs/update_home.ts
@@ -13,7 +13,7 @@ import { findActiveUser } from '../services/user'
 import { lanaugeToCode } from '../utils/helpers'
 import { logError, logger } from '../utils/logger'
 
-export const UPDATE_HOME_JOB = 'UPDATE_HOME_JOB'
+export const UPDATE_HOME_JOB = 'update-home'
 
 export interface UpdateHomeJobData {
   userId: string
diff --git a/packages/api/src/jobs/upload_content.ts b/packages/api/src/jobs/upload_content.ts
index 2389aad4e..4ffb5bdb9 100644
--- a/packages/api/src/jobs/upload_content.ts
+++ b/packages/api/src/jobs/upload_content.ts
@@ -4,7 +4,7 @@ import { logger } from '../utils/logger'
 import { htmlToHighlightedMarkdown, htmlToMarkdown } from '../utils/parser'
 import { isFileExists, uploadToBucket } from '../utils/uploads'
 
-export const UPLOAD_CONTENT_JOB = 'UPLOAD_CONTENT_JOB'
+export const UPLOAD_CONTENT_JOB = 'upload-content'
 
 export type ContentFormat =
   | 'markdown'

From a580fd7ebfde815a011a332d2f8fe67ffbe628cb Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Wed, 12 Jun 2024 17:06:43 +0800
Subject: [PATCH 3/9] catch sync reading progress query error

---
 packages/api/src/jobs/sync_read_positions.ts | 8 ++++++--
 packages/api/src/jobs/update_home.ts         | 2 +-
 2 files changed, 7 insertions(+), 3 deletions(-)

diff --git a/packages/api/src/jobs/sync_read_positions.ts b/packages/api/src/jobs/sync_read_positions.ts
index 26ec0b639..756d6429d 100644
--- a/packages/api/src/jobs/sync_read_positions.ts
+++ b/packages/api/src/jobs/sync_read_positions.ts
@@ -6,8 +6,8 @@ import {
   fetchCachedReadingPositionsAndMembers,
   reduceCachedReadingPositionMembers,
 } from '../services/cached_reading_position'
-import { logger } from '../utils/logger'
 import { updateLibraryItemReadingProgress } from '../services/library_item'
+import { logger } from '../utils/logger'
 
 export const SYNC_READ_POSITIONS_JOB_NAME = 'sync-read-positions'
 
@@ -86,6 +86,10 @@ export const syncReadPositionsJob = async (_data: any) => {
 
   const updates = getSyncUpdatesIterator(redis)
   for await (const value of updates) {
-    await syncReadPosition(value)
+    try {
+      await syncReadPosition(value)
+    } catch (error) {
+      logger.error('error syncing reading position', { error, value })
+    }
   }
 }
diff --git a/packages/api/src/jobs/update_home.ts b/packages/api/src/jobs/update_home.ts
index 72e39f3d5..8994fee20 100644
--- a/packages/api/src/jobs/update_home.ts
+++ b/packages/api/src/jobs/update_home.ts
@@ -438,7 +438,7 @@ const mixHomeItems = (
 
 // use prometheus to monitor the latency of each step
 const latency = new client.Histogram({
-  name: 'update_home_latency',
+  name: 'omnivore_update_home_latency',
   help: 'Latency of update home job',
   labelNames: ['step'],
   buckets: [0.1, 0.5, 1, 2, 5, 10],

From f29592d67147aa754db7e457661a49b49617f3fa Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Wed, 12 Jun 2024 18:37:58 +0800
Subject: [PATCH 4/9] worker can concurrently process 10 jobs

---
 packages/api/src/queue-processor.ts | 1 +
 1 file changed, 1 insertion(+)

diff --git a/packages/api/src/queue-processor.ts b/packages/api/src/queue-processor.ts
index 4c4643070..5e9aec3a9 100644
--- a/packages/api/src/queue-processor.ts
+++ b/packages/api/src/queue-processor.ts
@@ -227,6 +227,7 @@ export const createWorker = (connection: ConnectionOptions) =>
       connection,
       autorun: true, // start processing jobs immediately
       lockDuration: 60_000, // 1 minute
+      concurrency: 10,
     }
   )
 

From fb5ef3422cb5e581c20ac55489a388471de9b186 Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Wed, 12 Jun 2024 19:09:40 +0800
Subject: [PATCH 5/9] commit the delete query for each user in the loop to
 reduce lock time

---
 .../0181.do.batch_delete_trash_items.sql      | 24 +++++++++++++++++--
 .../0181.undo.batch_delete_trash_items.sql    |  2 +-
 2 files changed, 23 insertions(+), 3 deletions(-)

diff --git a/packages/db/migrations/0181.do.batch_delete_trash_items.sql b/packages/db/migrations/0181.do.batch_delete_trash_items.sql
index 07b38d538..3657610ae 100755
--- a/packages/db/migrations/0181.do.batch_delete_trash_items.sql
+++ b/packages/db/migrations/0181.do.batch_delete_trash_items.sql
@@ -4,7 +4,9 @@
 
 BEGIN;
 
-CREATE OR REPLACE FUNCTION batch_delete_trash_items()
+CREATE OR REPLACE PROCEDURE batch_delete_trash_items(
+    minimum_items INT
+)
 RETURNS VOID AS $$
 DECLARE
     user_record RECORD;
@@ -17,14 +19,32 @@ DECLARE
         status = 'ACTIVE';
 BEGIN
     FOR user_record IN user_cursor LOOP
+        BEGIN;
+        
         -- For Row Level Security
         PERFORM omnivore.set_claims(user_record.id, 'omnivore_user');
 
+        -- keep the minimum number of items in trash
         DELETE FROM omnivore.library_item
         WHERE
             user_id = user_record.id
             AND state = 'DELETED'
-            AND deleted_at < NOW() - INTERVAL '14 days';
+            AND deleted_at < NOW() - INTERVAL '14 days'
+            AND id NOT IN (
+                SELECT
+                    id
+                FROM
+                    omnivore.library_item
+                WHERE
+                    user_id = user_record.id
+                    AND state = 'DELETED'
+                    AND deleted_at < NOW() - INTERVAL '14 days'
+                ORDER BY
+                    deleted_at DESC
+                LIMIT minimum_items
+            );
+
+        COMMIT;
     END LOOP;
 END;
 $$ LANGUAGE plpgsql;
diff --git a/packages/db/migrations/0181.undo.batch_delete_trash_items.sql b/packages/db/migrations/0181.undo.batch_delete_trash_items.sql
index 836f8e533..3851c057b 100755
--- a/packages/db/migrations/0181.undo.batch_delete_trash_items.sql
+++ b/packages/db/migrations/0181.undo.batch_delete_trash_items.sql
@@ -4,6 +4,6 @@
 
 BEGIN;
 
-DROP FUNCTION IF EXISTS batch_delete_trash_items();
+DROP PROCEDURE IF EXISTS batch_delete_trash_items();
 
 COMMIT;

From 46aa4055520a2ad0ecfe0c97896d140a66457070 Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Wed, 12 Jun 2024 19:13:09 +0800
Subject: [PATCH 6/9] remove timeout in highlight generator

---
 packages/api/src/utils/highlightGenerator.ts | 9 ---------
 1 file changed, 9 deletions(-)

diff --git a/packages/api/src/utils/highlightGenerator.ts b/packages/api/src/utils/highlightGenerator.ts
index 8f9da4e1f..4cc891f33 100644
--- a/packages/api/src/utils/highlightGenerator.ts
+++ b/packages/api/src/utils/highlightGenerator.ts
@@ -49,8 +49,6 @@ type FillNodeResponse = {
 }
 
 function getTextNodesBetween(rootNode: Node, startNode: Node, endNode: Node) {
-  const maxTime = 10_000 // 10 seconds
-  const start = Date.now()
   let textNodeStartingPoint = 0
   let articleText = ''
   let newParagraph = false
@@ -70,13 +68,6 @@ function getTextNodesBetween(rootNode: Node, startNode: Node, endNode: Node) {
   }
 
   function getTextNodes(node: Node) {
-    // If the function takes too long, throw an error
-    if (Date.now() - start > maxTime) {
-      const error = new Error('getTextNodes Timeout')
-      logger.error(error)
-      throw error
-    }
-
     if (!node) return
 
     if (node == startNode) {

From 71d07caf942c1b25e882e6b8c39fd1d3b8c3b1c4 Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Wed, 12 Jun 2024 19:27:42 +0800
Subject: [PATCH 7/9] run prune trash job at 3am UTC daily

---
 packages/api/src/jobs/prune_library_item.ts   |  0
 packages/api/src/jobs/prune_trash.ts          |  5 +++++
 packages/api/src/queue-processor.ts           | 15 ++++++++++++++
 packages/api/src/services/library_item.ts     |  3 +++
 packages/api/src/utils/createTask.ts          |  2 ++
 .../0181.do.batch_delete_trash_items.sql      | 20 ++-----------------
 .../0181.undo.batch_delete_trash_items.sql    |  2 +-
 7 files changed, 28 insertions(+), 19 deletions(-)
 delete mode 100644 packages/api/src/jobs/prune_library_item.ts
 create mode 100644 packages/api/src/jobs/prune_trash.ts

diff --git a/packages/api/src/jobs/prune_library_item.ts b/packages/api/src/jobs/prune_library_item.ts
deleted file mode 100644
index e69de29bb..000000000
diff --git a/packages/api/src/jobs/prune_trash.ts b/packages/api/src/jobs/prune_trash.ts
new file mode 100644
index 000000000..953c060be
--- /dev/null
+++ b/packages/api/src/jobs/prune_trash.ts
@@ -0,0 +1,5 @@
+import { pruneTrash } from '../services/library_item'
+
+export const PRUNE_TRASH_JOB = 'prune_trash'
+
+export const pruneTrashJob = async () => pruneTrash()
diff --git a/packages/api/src/queue-processor.ts b/packages/api/src/queue-processor.ts
index 5e9aec3a9..44df2e905 100644
--- a/packages/api/src/queue-processor.ts
+++ b/packages/api/src/queue-processor.ts
@@ -50,6 +50,7 @@ import {
   PROCESS_YOUTUBE_TRANSCRIPT_JOB_NAME,
   PROCESS_YOUTUBE_VIDEO_JOB_NAME,
 } from './jobs/process-youtube-video'
+import { pruneTrashJob, PRUNE_TRASH_JOB } from './jobs/prune_trash'
 import { refreshAllFeeds } from './jobs/rss/refreshAllFeeds'
 import { refreshFeed } from './jobs/rss/refreshFeed'
 import { savePageJob } from './jobs/save_page'
@@ -214,6 +215,8 @@ export const createWorker = (connection: ConnectionOptions) =>
             return scoreLibraryItem(job.data)
           case GENERATE_PREVIEW_CONTENT_JOB:
             return generatePreviewContent(job.data)
+          case PRUNE_TRASH_JOB:
+            return pruneTrashJob()
           default:
             logger.warning(`[queue-processor] unhandled job: ${job.name}`)
         }
@@ -248,6 +251,18 @@ const setupCronJobs = async () => {
       },
     }
   )
+
+  await queue.add(
+    PRUNE_TRASH_JOB,
+    {},
+    {
+      priority: getJobPriority(PRUNE_TRASH_JOB),
+      repeat: {
+        // daily at 3am
+        pattern: '0 3 * * *',
+      },
+    }
+  )
 }
 
 const main = async () => {
diff --git a/packages/api/src/services/library_item.ts b/packages/api/src/services/library_item.ts
index 8cc8914ec..7a7a898f3 100644
--- a/packages/api/src/services/library_item.ts
+++ b/packages/api/src/services/library_item.ts
@@ -1779,3 +1779,6 @@ export const downloadOriginalContent = async (
     })
   )
 }
+
+export const pruneTrash = async () =>
+  appDataSource.query(`CALL omnivore.batch_delete_trash_items();`)
diff --git a/packages/api/src/utils/createTask.ts b/packages/api/src/utils/createTask.ts
index 212628d38..f10971332 100644
--- a/packages/api/src/utils/createTask.ts
+++ b/packages/api/src/utils/createTask.ts
@@ -41,6 +41,7 @@ import {
   PROCESS_YOUTUBE_TRANSCRIPT_JOB_NAME,
   PROCESS_YOUTUBE_VIDEO_JOB_NAME,
 } from '../jobs/process-youtube-video'
+import { PRUNE_TRASH_JOB } from '../jobs/prune_trash'
 import {
   queueRSSRefreshFeedJob,
   REFRESH_ALL_FEEDS_JOB_NAME,
@@ -112,6 +113,7 @@ export const getJobPriority = (jobName: string): number => {
     case REFRESH_ALL_FEEDS_JOB_NAME:
     case THUMBNAIL_JOB:
     case GENERATE_PREVIEW_CONTENT_JOB:
+    case PRUNE_TRASH_JOB:
       return 100
 
     default:
diff --git a/packages/db/migrations/0181.do.batch_delete_trash_items.sql b/packages/db/migrations/0181.do.batch_delete_trash_items.sql
index 3657610ae..38a619469 100755
--- a/packages/db/migrations/0181.do.batch_delete_trash_items.sql
+++ b/packages/db/migrations/0181.do.batch_delete_trash_items.sql
@@ -4,9 +4,7 @@
 
 BEGIN;
 
-CREATE OR REPLACE PROCEDURE batch_delete_trash_items(
-    minimum_items INT
-)
+CREATE OR REPLACE PROCEDURE omnivore.batch_delete_trash_items()
 RETURNS VOID AS $$
 DECLARE
     user_record RECORD;
@@ -24,25 +22,11 @@ BEGIN
         -- For Row Level Security
         PERFORM omnivore.set_claims(user_record.id, 'omnivore_user');
 
-        -- keep the minimum number of items in trash
         DELETE FROM omnivore.library_item
         WHERE
             user_id = user_record.id
             AND state = 'DELETED'
-            AND deleted_at < NOW() - INTERVAL '14 days'
-            AND id NOT IN (
-                SELECT
-                    id
-                FROM
-                    omnivore.library_item
-                WHERE
-                    user_id = user_record.id
-                    AND state = 'DELETED'
-                    AND deleted_at < NOW() - INTERVAL '14 days'
-                ORDER BY
-                    deleted_at DESC
-                LIMIT minimum_items
-            );
+            AND deleted_at < NOW() - INTERVAL '14 days';
 
         COMMIT;
     END LOOP;
diff --git a/packages/db/migrations/0181.undo.batch_delete_trash_items.sql b/packages/db/migrations/0181.undo.batch_delete_trash_items.sql
index 3851c057b..3734a1db1 100755
--- a/packages/db/migrations/0181.undo.batch_delete_trash_items.sql
+++ b/packages/db/migrations/0181.undo.batch_delete_trash_items.sql
@@ -4,6 +4,6 @@
 
 BEGIN;
 
-DROP PROCEDURE IF EXISTS batch_delete_trash_items();
+DROP PROCEDURE IF EXISTS omnivore.batch_delete_trash_items();
 
 COMMIT;

From 30adbb735f13e12dfc3e5c6972fea018d4cf35f7 Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Wed, 12 Jun 2024 19:41:07 +0800
Subject: [PATCH 8/9] fix typo in sql

---
 .../0181.do.batch_delete_trash_items.sql      | 25 ++++++++++---------
 1 file changed, 13 insertions(+), 12 deletions(-)

diff --git a/packages/db/migrations/0181.do.batch_delete_trash_items.sql b/packages/db/migrations/0181.do.batch_delete_trash_items.sql
index 38a619469..38518b6b6 100755
--- a/packages/db/migrations/0181.do.batch_delete_trash_items.sql
+++ b/packages/db/migrations/0181.do.batch_delete_trash_items.sql
@@ -5,7 +5,7 @@
 BEGIN;
 
 CREATE OR REPLACE PROCEDURE omnivore.batch_delete_trash_items()
-RETURNS VOID AS $$
+LANGUAGE plpgsql AS $$
 DECLARE
     user_record RECORD;
     user_cursor CURSOR FOR
@@ -17,20 +17,21 @@ DECLARE
         status = 'ACTIVE';
 BEGIN
     FOR user_record IN user_cursor LOOP
-        BEGIN;
+        BEGIN
         
-        -- For Row Level Security
-        PERFORM omnivore.set_claims(user_record.id, 'omnivore_user');
+            -- For Row Level Security
+            PERFORM omnivore.set_claims(user_record.id, 'omnivore_user');
 
-        DELETE FROM omnivore.library_item
-        WHERE
-            user_id = user_record.id
-            AND state = 'DELETED'
-            AND deleted_at < NOW() - INTERVAL '14 days';
+            DELETE FROM omnivore.library_item
+            WHERE
+                user_id = user_record.id
+                AND state = 'DELETED'
+                AND deleted_at < NOW() - INTERVAL '14 days';
 
-        COMMIT;
+            COMMIT;
+        END;
     END LOOP;
-END;
-$$ LANGUAGE plpgsql;
+END
+$$;
 
 COMMIT;

From 9ef91ff1cae5a9b552477a2460876b64d1b2b7d3 Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Thu, 13 Jun 2024 14:48:14 +0800
Subject: [PATCH 9/9] add an REST API to trigger the prune trash job

---
 packages/api/src/jobs/prune_trash.ts          | 13 ++++-
 packages/api/src/queue-processor.ts           | 14 +----
 packages/api/src/routers/svc/links.ts         | 53 ++++++-------------
 packages/api/src/services/library_item.ts     | 14 -----
 packages/api/src/utils/createTask.ts          | 19 +++++++
 .../0181.do.batch_delete_trash_items.sql      |  6 ++-
 6 files changed, 50 insertions(+), 69 deletions(-)

diff --git a/packages/api/src/jobs/prune_trash.ts b/packages/api/src/jobs/prune_trash.ts
index 953c060be..55f54694c 100644
--- a/packages/api/src/jobs/prune_trash.ts
+++ b/packages/api/src/jobs/prune_trash.ts
@@ -1,5 +1,14 @@
-import { pruneTrash } from '../services/library_item'
+import { appDataSource } from '../data_source'
 
 export const PRUNE_TRASH_JOB = 'prune_trash'
 
-export const pruneTrashJob = async () => pruneTrash()
+interface PruneTrashJobData {
+  numDays: number
+}
+
+export const pruneTrashJob = async (jobData: PruneTrashJobData) => {
+  // call the stored procedure to delete trash items older than {numDays} days
+  await appDataSource.query(
+    `CALL omnivore.batch_delete_trash_items(${jobData.numDays});`
+  )
+}
diff --git a/packages/api/src/queue-processor.ts b/packages/api/src/queue-processor.ts
index 44df2e905..00440da53 100644
--- a/packages/api/src/queue-processor.ts
+++ b/packages/api/src/queue-processor.ts
@@ -216,7 +216,7 @@ export const createWorker = (connection: ConnectionOptions) =>
           case GENERATE_PREVIEW_CONTENT_JOB:
             return generatePreviewContent(job.data)
           case PRUNE_TRASH_JOB:
-            return pruneTrashJob()
+            return pruneTrashJob(job.data)
           default:
             logger.warning(`[queue-processor] unhandled job: ${job.name}`)
         }
@@ -251,18 +251,6 @@ const setupCronJobs = async () => {
       },
     }
   )
-
-  await queue.add(
-    PRUNE_TRASH_JOB,
-    {},
-    {
-      priority: getJobPriority(PRUNE_TRASH_JOB),
-      repeat: {
-        // daily at 3am
-        pattern: '0 3 * * *',
-      },
-    }
-  )
 }
 
 const main = async () => {
diff --git a/packages/api/src/routers/svc/links.ts b/packages/api/src/routers/svc/links.ts
index f247b6ffa..bd35351cc 100644
--- a/packages/api/src/routers/svc/links.ts
+++ b/packages/api/src/routers/svc/links.ts
@@ -2,12 +2,10 @@
 /* eslint-disable @typescript-eslint/no-unsafe-assignment */
 /* eslint-disable @typescript-eslint/explicit-module-boundary-types */
 import express from 'express'
-import { LessThan } from 'typeorm'
-import { LibraryItemState } from '../../entity/library_item'
 import { readPushSubscription } from '../../pubsub'
 import { userRepository } from '../../repository/user'
 import { createPageSaveRequest } from '../../services/create_page_save_request'
-import { deleteLibraryItemsByAdmin } from '../../services/library_item'
+import { enqueuePruneTrashJob } from '../../utils/createTask'
 import { logger } from '../../utils/logger'
 
 interface CreateLinkRequestMessage {
@@ -16,28 +14,7 @@ interface CreateLinkRequestMessage {
 }
 
 type PruneMessage = {
-  expireInDays: number
-  folder?: string
-  state?: LibraryItemState
-}
-
-const isPruneMessage = (obj: any): obj is PruneMessage => 'expireInDays' in obj
-
-const getPruneMessage = (msgStr: string): PruneMessage => {
-  try {
-    const obj = JSON.parse(msgStr) as unknown
-    if (isPruneMessage(obj)) {
-      return obj
-    }
-  } catch (err) {
-    logger.error('error deserializing event: ', { msgStr, err })
-  }
-
-  // default to prune following folder items older than 30 days
-  return {
-    folder: 'following',
-    expireInDays: 30,
-  }
+  ttlInDays?: number
 }
 
 export function linkServiceRouter() {
@@ -84,28 +61,28 @@ export function linkServiceRouter() {
     }
   })
 
-  router.post('/prune', async (req, res) => {
+  router.post('/pruneTrash', async (req, res) => {
     const { message: msgStr, expired } = readPushSubscription(req)
 
-    if (!msgStr) {
-      return res.status(200).send('Bad Request')
-    }
-
     if (expired) {
       logger.info('discarding expired message')
       return res.status(200).send('Expired')
     }
 
-    const pruneMessage = getPruneMessage(msgStr)
-    const expireTime = pruneMessage.expireInDays * 1000 * 60 * 60 * 24 // convert days to milliseconds
+    // default to prune trash items older than 14 days
+    let ttlInDays = 14
+
+    if (msgStr) {
+      const pruneMessage = JSON.parse(msgStr) as PruneMessage
+
+      if (pruneMessage.ttlInDays) {
+        ttlInDays = pruneMessage.ttlInDays
+      }
+    }
 
     try {
-      const result = await deleteLibraryItemsByAdmin({
-        folder: pruneMessage.folder,
-        state: pruneMessage.state,
-        updatedAt: LessThan(new Date(Date.now() - expireTime)),
-      })
-      logger.info('prune result', result)
+      const job = await enqueuePruneTrashJob(ttlInDays)
+      logger.info('enqueue prune trash job', { id: job?.id })
 
       return res.sendStatus(200)
     } catch (error) {
diff --git a/packages/api/src/services/library_item.ts b/packages/api/src/services/library_item.ts
index 7a7a898f3..2727cedf5 100644
--- a/packages/api/src/services/library_item.ts
+++ b/packages/api/src/services/library_item.ts
@@ -1354,17 +1354,6 @@ export const deleteLibraryItemsByUserId = async (userId: string) => {
   )
 }
 
-export const deleteLibraryItemsByAdmin = async (
-  criteria: FindOptionsWhere<LibraryItem>
-) => {
-  return authTrx(
-    async (tx) => tx.withRepository(libraryItemRepository).delete(criteria),
-    undefined,
-    undefined,
-    'admin'
-  )
-}
-
 export const batchDelete = async (criteria: FindOptionsWhere<LibraryItem>) => {
   const batchSize = 1000
 
@@ -1779,6 +1768,3 @@ export const downloadOriginalContent = async (
     })
   )
 }
-
-export const pruneTrash = async () =>
-  appDataSource.query(`CALL omnivore.batch_delete_trash_items();`)
diff --git a/packages/api/src/utils/createTask.ts b/packages/api/src/utils/createTask.ts
index f10971332..2957926fd 100644
--- a/packages/api/src/utils/createTask.ts
+++ b/packages/api/src/utils/createTask.ts
@@ -1053,4 +1053,23 @@ export const enqueueGeneratePreviewContentJob = async (
   )
 }
 
+export const enqueuePruneTrashJob = async (numDays: number) => {
+  const queue = await getBackendQueue()
+  if (!queue) {
+    return undefined
+  }
+
+  return queue.add(
+    PRUNE_TRASH_JOB,
+    { numDays },
+    {
+      jobId: `${PRUNE_TRASH_JOB}_${numDays}_${JOB_VERSION}`,
+      removeOnComplete: true,
+      removeOnFail: true,
+      priority: getJobPriority(PRUNE_TRASH_JOB),
+      attempts: 3,
+    }
+  )
+}
+
 export default createHttpTaskWithToken
diff --git a/packages/db/migrations/0181.do.batch_delete_trash_items.sql b/packages/db/migrations/0181.do.batch_delete_trash_items.sql
index 38518b6b6..afcdcc880 100755
--- a/packages/db/migrations/0181.do.batch_delete_trash_items.sql
+++ b/packages/db/migrations/0181.do.batch_delete_trash_items.sql
@@ -4,7 +4,9 @@
 
 BEGIN;
 
-CREATE OR REPLACE PROCEDURE omnivore.batch_delete_trash_items()
+CREATE OR REPLACE PROCEDURE omnivore.batch_delete_trash_items(
+    num_days INT
+)
 LANGUAGE plpgsql AS $$
 DECLARE
     user_record RECORD;
@@ -26,7 +28,7 @@ BEGIN
             WHERE
                 user_id = user_record.id
                 AND state = 'DELETED'
-                AND deleted_at < NOW() - INTERVAL '14 days';
+                AND deleted_at < NOW() - INTERVAL '1 day' * num_days;
 
             COMMIT;
         END;