add folder expiration jobs

packages/api/src/entity/folder_policy.ts

@@ -10,8 +10,8 @@ import {
 import { User } from './user'
 
 export enum FolderPolicyAction {
-  DELETE = 'DELETE',
-  ARCHIVE = 'ARCHIVE',
+  Delete = 'DELETE',
+  Archive = 'ARCHIVE',
 }
 
 @Entity({ name: 'folder_policy' })

packages/api/src/jobs/folder/expire.ts

@@ -0,0 +1,50 @@
+import { FolderPolicyAction } from '../../entity/folder_policy'
+import { BulkActionType } from '../../generated/graphql'
+import { findFolderPolicyById } from '../../services/folder_policy'
+import { batchUpdateLibraryItems } from '../../services/library_item'
+import { logger } from '../../utils/logger'
+
+export const EXPIRE_FOLDER_JOB_NAME = 'EXPIRE_FOLDER_JOB'
+
+interface ExpireFolderJobData {
+  userId: string
+  folderPolicyId: string
+}
+
+export const expireFolderJob = async (data: ExpireFolderJobData) => {
+  const { userId, folderPolicyId } = data
+
+  const policy = await findFolderPolicyById(userId, folderPolicyId)
+  if (!policy) {
+    logger.error('Policy not found')
+    return
+  }
+
+  logger.info(`Expiring items for policy ${policy.id}`)
+
+  const getBulkActionType = (action: FolderPolicyAction) => {
+    switch (action) {
+      case FolderPolicyAction.Archive:
+        return BulkActionType.Archive
+      case FolderPolicyAction.Delete:
+        return BulkActionType.Delete
+      default:
+        logger.error('Unsupported action')
+        throw new Error('Unsupported action')
+    }
+  }
+
+  const action = getBulkActionType(policy.action)
+  const savedAfter = new Date(
+    Date.now() - policy.afterDays * 24 * 60 * 60 * 1000
+  )
+
+  await batchUpdateLibraryItems(
+    action,
+    {
+      useFolders: true,
+      query: `in:${policy.folder} saved:<${savedAfter.toISOString()}`,
+    },
+    userId
+  )
+}

packages/api/src/jobs/folder/expire_all.ts

@@ -0,0 +1,20 @@
+import { findFolderPolicies } from '../../services/folder_policy'
+import { enqueueExpireFolderJob } from '../../utils/createTask'
+import { logError } from '../../utils/logger'
+
+export const EXPIRE_ALL_FOLDERS_JOB_NAME = 'EXPIRE_ALL_FOLDERS_JOB'
+
+export const expireAllFoldersJob = async () => {
+  const policies = await findFolderPolicies()
+
+  // sequentially enqueues a job to expire items for each policy
+  for (const policy of policies) {
+    try {
+      await enqueueExpireFolderJob(policy.userId, policy.id)
+    } catch (error) {
+      logError(error)
+
+      continue
+    }
+  }
+}

packages/api/src/queue-processor.ts

@@ -32,6 +32,11 @@ import {
 } from './jobs/email/inbound_emails'
 import { sendEmailJob, SEND_EMAIL_JOB } from './jobs/email/send_email'
 import { findThumbnail, THUMBNAIL_JOB } from './jobs/find_thumbnail'
+import { expireFolderJob, EXPIRE_FOLDER_JOB_NAME } from './jobs/folder/expire'
+import {
+  expireAllFoldersJob,
+  EXPIRE_ALL_FOLDERS_JOB_NAME,
+} from './jobs/folder/expire_all'
 import {
   generatePreviewContent,
   GENERATE_PREVIEW_CONTENT_JOB,
@@ -217,6 +222,10 @@ export const createWorker = (connection: ConnectionOptions) =>
             return generatePreviewContent(job.data)
           case PRUNE_TRASH_JOB:
             return pruneTrashJob(job.data)
+          case EXPIRE_ALL_FOLDERS_JOB_NAME:
+            return expireAllFoldersJob()
+          case EXPIRE_FOLDER_JOB_NAME:
+            return expireFolderJob(job.data)
           default:
             logger.warning(`[queue-processor] unhandled job: ${job.name}`)
         }

packages/api/src/utils/createTask.ts

@@ -29,6 +29,8 @@ import { BulkActionData, BULK_ACTION_JOB_NAME } from '../jobs/bulk_action'
 import { CallWebhookJobData, CALL_WEBHOOK_JOB_NAME } from '../jobs/call_webhook'
 import { SendEmailJobData, SEND_EMAIL_JOB } from '../jobs/email/send_email'
 import { THUMBNAIL_JOB } from '../jobs/find_thumbnail'
+import { EXPIRE_FOLDER_JOB_NAME } from '../jobs/folder/expire'
+import { EXPIRE_ALL_FOLDERS_JOB_NAME } from '../jobs/folder/expire_all'
 import { GENERATE_PREVIEW_CONTENT_JOB } from '../jobs/generate_preview_content'
 import { EXPORT_ALL_ITEMS_JOB_NAME } from '../jobs/integration/export_all_items'
 import {
@@ -114,6 +116,8 @@ export const getJobPriority = (jobName: string): number => {
     case THUMBNAIL_JOB:
     case GENERATE_PREVIEW_CONTENT_JOB:
     case PRUNE_TRASH_JOB:
+    case EXPIRE_ALL_FOLDERS_JOB_NAME:
+    case EXPIRE_FOLDER_JOB_NAME:
       return 100
 
     default:
@@ -1072,4 +1076,45 @@ export const enqueuePruneTrashJob = async (numDays: number) => {
   )
 }
 
+export const enqueueExpireAllFoldersJob = async () => {
+  const queue = await getBackendQueue()
+  if (!queue) {
+    return undefined
+  }
+
+  return queue.add(
+    EXPIRE_ALL_FOLDERS_JOB_NAME,
+    {},
+    {
+      jobId: `${EXPIRE_ALL_FOLDERS_JOB_NAME}_${JOB_VERSION}`,
+      removeOnComplete: true,
+      removeOnFail: true,
+      priority: getJobPriority(EXPIRE_ALL_FOLDERS_JOB_NAME),
+      attempts: 1,
+    }
+  )
+}
+
+export const enqueueExpireFolderJob = async (
+  userId: string,
+  folderPolicyId: string
+) => {
+  const queue = await getBackendQueue()
+  if (!queue) {
+    return undefined
+  }
+
+  return queue.add(
+    EXPIRE_FOLDER_JOB_NAME,
+    { userId, folderPolicyId },
+    {
+      jobId: `${EXPIRE_FOLDER_JOB_NAME}_${folderPolicyId}_${JOB_VERSION}`,
+      removeOnComplete: true,
+      removeOnFail: true,
+      priority: getJobPriority(EXPIRE_FOLDER_JOB_NAME),
+      attempts: 3,
+    }
+  )
+}
+
 export default createHttpTaskWithToken

packages/api/test/resolvers/folder_policy.test.ts

@@ -55,14 +55,14 @@ describe('Folder Policy API', () => {
       const existingPolicy = await createFolderPolicy({
         userId: loginUser.id,
         folder: 'inbox',
-        action: FolderPolicyAction.ARCHIVE,
+        action: FolderPolicyAction.Archive,
         afterDays: 30,
         minimumItems: 10,
       })
       const existingPolicy1 = await createFolderPolicy({
         userId: loginUser.id,
         folder: 'following',
-        action: FolderPolicyAction.ARCHIVE,
+        action: FolderPolicyAction.Archive,
         afterDays: 30,
         minimumItems: 10,
       })
@@ -102,7 +102,7 @@ describe('Folder Policy API', () => {
     it('should create a folder policy', async () => {
       const input = {
         folder: 'test-folder',
-        action: FolderPolicyAction.ARCHIVE,
+        action: FolderPolicyAction.Archive,
         afterDays: 30,
         minimumItems: 10,
       }
@@ -129,7 +129,7 @@ describe('Folder Policy API', () => {
       existingPolicy = await createFolderPolicy({
         userId: loginUser.id,
         folder: 'test-folder',
-        action: FolderPolicyAction.ARCHIVE,
+        action: FolderPolicyAction.Archive,
         afterDays: 30,
         minimumItems: 10,
       })
@@ -161,7 +161,7 @@ describe('Folder Policy API', () => {
     it('should update a folder policy', async () => {
       const input = {
         id: existingPolicy.id,
-        action: FolderPolicyAction.DELETE,
+        action: FolderPolicyAction.Delete,
         afterDays: 30,
         minimumItems: 10,
       }
@@ -186,7 +186,7 @@ describe('Folder Policy API', () => {
       existingPolicy = await createFolderPolicy({
         userId: loginUser.id,
         folder: 'test-folder',
-        action: FolderPolicyAction.ARCHIVE,
+        action: FolderPolicyAction.Archive,
         afterDays: 30,
         minimumItems: 10,
       })