content-fetch: mark final-attempt failures as FAILED

Ensure items don"t remain stuck in PROCESSING by marking the backing library_item as FAILED when content-fetch exhausts retries (or hits a blocked domain).

packages/api/src/jobs/save_page.ts

@@ -6,10 +6,12 @@ import {
   ArticleSavingRequestStatus,
   CreateLabelInput,
 } from '../generated/graphql'
+import { LibraryItemState } from '../entity/library_item'
 import { redisDataSource } from '../redis_data_source'
 import { userRepository } from '../repository/user'
 import { saveFile } from '../services/save_file'
 import { savePage } from '../services/save_page'
+import { updateLibraryItem } from '../services/library_item'
 import { uploadFile } from '../services/upload_file'
 import { logError, logger } from '../utils/logger'
 import {
@@ -285,6 +287,21 @@ export const savePageJob = async (data: Data, attemptsMade: number) => {
     isSaved = true
   } catch (e) {
     logError(e)
+    try {
+      await updateLibraryItem(
+        articleSavingRequestId,
+        {
+          state: LibraryItemState.Failed,
+        },
+        userId
+      )
+    } catch (updateError) {
+      logger.error('Failed to mark library item as FAILED', {
+        userId,
+        articleSavingRequestId,
+        updateError,
+      })
+    }
 
     throw e
   } finally {

packages/content-fetch/src/request_handler.ts

@@ -17,7 +17,8 @@ interface UserConfig {
 export interface JobData {
   url: string
   userId?: string
-  saveRequestId: string
+  // Back-compat: older payloads used saveRequestId + userId; newer payloads use users[].
+  saveRequestId?: string
   state?: string
   labels?: string[]
   source?: string
@@ -34,7 +35,7 @@ export interface JobData {
 
 interface LogRecord {
   url: string
-  articleSavingRequestId: string
+  articleSavingRequestId?: string
   labels: {
     source: string
   }
@@ -75,9 +76,29 @@ const signToken = promisify(jwt.sign)
 const IMPORTER_METRICS_COLLECTOR_URL =
   process.env.IMPORTER_METRICS_COLLECTOR_URL
 const JWT_SECRET = process.env.JWT_SECRET
+const REST_BACKEND_ENDPOINT = process.env.REST_BACKEND_ENDPOINT
+
+const API_GRAPHQL_ENDPOINT =
+  process.env.API_GRAPHQL_ENDPOINT ||
+  (REST_BACKEND_ENDPOINT
+    ? REST_BACKEND_ENDPOINT.replace(/\/api\/?$/i, '/api/graphql')
+    : undefined)
 
 const MAX_IMPORT_ATTEMPTS = 1
 
+const UPDATE_PAGE_MUTATION = `
+  mutation UpdatePage($input: UpdatePageInput!) {
+    updatePage(input: $input) {
+      ... on UpdatePageSuccess {
+        updatedPage { id }
+      }
+      ... on UpdatePageError {
+        errorCodes
+      }
+    }
+  }
+`
+
 const uploadToBucket = async (filePath: string, data: string) => {
   await storage
     .bucket(bucketName)
@@ -219,6 +240,58 @@ const sendImportStatusUpdate = async (
   }
 }
 
+export const markFetchFailure = async (
+  users: UserConfig[],
+  errorMessage: string
+) => {
+  if (!JWT_SECRET || !API_GRAPHQL_ENDPOINT) {
+    console.error('JWT_SECRET or API_GRAPHQL_ENDPOINT is not set, cannot mark failure', {
+      hasJwtSecret: !!JWT_SECRET,
+      apiGraphqlEndpoint: API_GRAPHQL_ENDPOINT,
+    })
+    return
+  }
+
+  await Promise.all(
+    users.map(async (user) => {
+      try {
+        const auth = (await signToken(
+          { uid: user.id },
+          JWT_SECRET
+        )) as string
+
+        await axios.post(
+          API_GRAPHQL_ENDPOINT,
+          {
+            query: UPDATE_PAGE_MUTATION,
+            variables: {
+              input: {
+                pageId: user.libraryItemId,
+                state: 'FAILED',
+                // Keep existing savedAt/publishedAt/title/etc.
+              },
+            },
+          },
+          {
+            headers: {
+              'Omnivore-Authorization': auth,
+              'Content-Type': 'application/json',
+            },
+            timeout: 5000,
+          }
+        )
+      } catch (e) {
+        console.error('Failed to mark fetch failure', {
+          userId: user.id,
+          libraryItemId: user.libraryItemId,
+          errorMessage,
+        })
+        console.error(e)
+      }
+    })
+  )
+}
+
 export const processFetchContentJob = async (
   redisDataSource: RedisDataSource,
   data: JobData,
@@ -235,9 +308,9 @@ export const processFetchContentJob = async (
       {
         id: userId,
         folder: data.folder,
-        libraryItemId: data.saveRequestId,
+        libraryItemId: data.saveRequestId || '',
       },
-    ]
+    ].filter((u) => !!u.libraryItemId)
   }
   const articleSavingRequestId = data.saveRequestId
   const state = data.state
@@ -278,6 +351,10 @@ export const processFetchContentJob = async (
       console.log('domain is blocked', domain)
       logRecord.error = 'domain is blocked'
 
+      if (users.length) {
+        await markFetchFailure(users, logRecord.error)
+      }
+
       return
     }
 

packages/content-fetch/src/worker.ts

@@ -1,6 +1,6 @@
 import { RedisDataSource } from '@omnivore/utils'
 import { Job, Queue, RedisClient, Worker } from 'bullmq'
-import { JobData, processFetchContentJob } from './request_handler'
+import { JobData, markFetchFailure, processFetchContentJob } from './request_handler'
 
 export const QUEUE = 'omnivore-content-fetch-queue'
 
@@ -34,8 +34,31 @@ export const createWorker = (
   const worker = new Worker(
     queueName,
     async (job: Job<JobData>) => {
-      // process the job
-      await processFetchContentJob(redisDataSource, job.data, job.attemptsMade)
+      const maxAttempts = job.opts.attempts ?? 1
+      try {
+        // process the job
+        await processFetchContentJob(redisDataSource, job.data, job.attemptsMade)
+      } catch (e) {
+        const lastAttempt = job.attemptsMade + 1 >= maxAttempts
+        if (lastAttempt) {
+          const errorMessage = e instanceof Error ? e.message : 'unknown error'
+          let users = job.data.users || []
+          if (job.data.userId) {
+            // Back-compat payload path
+            users = [
+              {
+                id: job.data.userId,
+                folder: job.data.folder,
+                libraryItemId: job.data.saveRequestId || '',
+              },
+            ].filter((u) => u.libraryItemId)
+          }
+          if (users.length) {
+            await markFetchFailure(users, errorMessage)
+          }
+        }
+        throw e
+      }
     },
     {
       connection: redisDataSource.queueRedisClient,

packages/content-fetch/start.sh

@@ -1,3 +1,3 @@
 #!/bin/sh
 cat hosts >> /etc/hosts
-yarn workspace @omnivore/content-fetch start
+yarn workspace @omnivore/content-fetch start