prehistoric-systems/omnivore
1
0
Fork 0
mirror of https://github.com/omnivore-app/omnivore.git synced 2026-03-11 08:54:26 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #3398 from omnivore-app/fix/rss-logging

Browse source 
More logging for refreshing feeds
This commit is contained in:
Jackson Harper 2024-01-22 12:24:40 +08:00 committed by GitHub
commit c939e9eb8e
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
37 changed files with 300 additions and 326 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

44
packages/api/src/jobs/rss/refreshAllFeeds.ts
View file

@ -1,12 +1,24 @@
import { Job, Queue } from 'bullmq'
import { DataSource } from 'typeorm'
import { QUEUE_NAME } from '../../queue-processor'
import { QUEUE_NAME, getBackendQueue } from '../../queue-processor'
import { redisDataSource } from '../../redis_data_source'
import { RssSubscriptionGroup } from '../../utils/createTask'
import { stringToHash } from '../../utils/helpers'
import { validateUrl } from '../../services/create_page_save_request'
import { v4 as uuid } from 'uuid'
export type RSSRefreshContext = {
type: 'all' | 'user-added'
refreshID: string
startedAt: string
}
export const refreshAllFeeds = async (db: DataSource): Promise<boolean> => {
const refreshContext = {
type: 'all',
refreshID: uuid(),
startedAt: new Date().toISOString(),
} as RSSRefreshContext
const subscriptionGroups = (await db.createEntityManager().query(
`
SELECT
@ -30,19 +42,31 @@ export const refreshAllFeeds = async (db: DataSource): Promise<boolean> => {
['RSS', 'ACTIVE', 'following']
)) as RssSubscriptionGroup[]
console.log(`rss: checking ${subscriptionGroups.length}`, { refreshContext })
for (const group of subscriptionGroups) {
try {
await updateSubscriptionGroup(group)
await updateSubscriptionGroup(group, refreshContext)
} catch (err) {
// we don't want to fail the whole job if one subscription group fails
console.error('error updating subscription group')
}
}
const finishTime = new Date()
console.log(
`rss: finished queuing subscription groups at ${finishTime.toISOString()}`,
{
refreshContext,
}
)
return true
}
const updateSubscriptionGroup = async (group: RssSubscriptionGroup) => {
const updateSubscriptionGroup = async (
group: RssSubscriptionGroup,
refreshContext: RSSRefreshContext
) => {
let feedURL = group.url
const userList = JSON.stringify(group.userIds.sort())
if (!feedURL) {
@ -63,6 +87,7 @@ const updateSubscriptionGroup = async (group: RssSubscriptionGroup) => {
userList
)}`
const payload = {
refreshContext,
subscriptionIds: group.subscriptionIds,
feedUrl: group.url,
lastFetchedTimestamps: group.fetchedDates.map(
@ -80,17 +105,8 @@ const updateSubscriptionGroup = async (group: RssSubscriptionGroup) => {
await queueRSSRefreshFeedJob(jobid, payload)
}
const createBackendQueue = (): Queue | undefined => {
if (!redisDataSource.workerRedisClient) {
throw new Error('Can not create queues, redis is not initialized')
}
return new Queue(QUEUE_NAME, {
connection: redisDataSource.workerRedisClient,
})
}
export const queueRSSRefreshAllFeedsJob = async () => {
const queue = createBackendQueue()
const queue = await getBackendQueue()
if (!queue) {
return false
}
@ -110,7 +126,7 @@ export const queueRSSRefreshFeedJob = async (
payload: any,
options = { priority: 'high' as QueuePriority }
): Promise<Job | undefined> => {
const queue = createBackendQueue()
const queue = await getBackendQueue()
if (!queue) {
return undefined
}

102
packages/api/src/jobs/rss/refreshFeed.ts
View file

@ -7,6 +7,8 @@ import { promisify } from 'util'
import { env } from '../../env'
import { redisDataSource } from '../../redis_data_source'
import createHttpTaskWithToken from '../../utils/createTask'
import { RSSRefreshContext } from './refreshAllFeeds'
import { updateSubscription } from '../../services/update_subscription'
type FolderType = 'following' | 'inbox'
@ -19,6 +21,7 @@ interface RefreshFeedRequest {
userIds: string[]
fetchContents: boolean[]
folders: FolderType[]
refreshContext?: RSSRefreshContext
}
export const isRefreshFeedRequest = (data: any): data is RefreshFeedRequest => {
@ -68,8 +71,6 @@ interface FetchContentTask {
item: RssFeedItem
}
const fetchContentTasks = new Map<string, FetchContentTask>() // url -> FetchContentTask
export const isOldItem = (item: RssFeedItem, lastFetchedAt: number) => {
// existing items and items that were published before 24h
const publishedAt = item.isoDate ? new Date(item.isoDate) : new Date()
@ -204,75 +205,6 @@ const parseFeed = async (url: string, content: string) => {
}
}
const sendUpdateSubscriptionMutation = async (
userId: string,
subscriptionId: string,
lastFetchedAt: Date,
lastFetchedChecksum: string,
scheduledAt: Date
) => {
if (!process.env.INTERNAL_API_URL || !env.server.jwtSecret) {
throw new Error(
'Can not send update subscription, environment not configured.'
)
}
const JWT_SECRET = env.server.jwtSecret
const REST_BACKEND_ENDPOINT = `${process.env.INTERNAL_API_URL}/api`
if (!JWT_SECRET || !REST_BACKEND_ENDPOINT) {
throw 'Environment not configured correctly'
}
const data = JSON.stringify({
query: `mutation UpdateSubscription($input: UpdateSubscriptionInput!){
updateSubscription(input:$input){
... on UpdateSubscriptionSuccess{
subscription{
id
lastFetchedAt
}
}
... on UpdateSubscriptionError{
errorCodes
}
}
}`,
variables: {
input: {
id: subscriptionId,
lastFetchedAt,
lastFetchedChecksum,
scheduledAt,
},
},
})
const auth = (await signToken({ uid: userId }, JWT_SECRET)) as string
try {
const response = await axios.post(
`${REST_BACKEND_ENDPOINT}/graphql`,
data,
{
headers: {
Cookie: `auth=${auth};`,
'Content-Type': 'application/json',
},
timeout: 30000, // 30s
}
)
/* eslint-disable @typescript-eslint/no-unsafe-member-access */
return !!response.data.data.updateSubscription.subscription
} catch (error) {
if (axios.isAxiosError(error)) {
console.error('update subscription mutation error', error.message)
} else {
console.error(error)
}
return false
}
}
const isItemRecentlySaved = async (userId: string, url: string) => {
const key = `recent-saved-item:${userId}:${url}`
try {
@ -286,6 +218,7 @@ const isItemRecentlySaved = async (userId: string, url: string) => {
}
const addFetchContentTask = (
fetchContentTasks: Map<string, FetchContentTask>,
userId: string,
folder: FolderType,
item: RssFeedItem
@ -305,6 +238,7 @@ const addFetchContentTask = (
}
const createTask = async (
fetchContentTasks: Map<string, FetchContentTask>,
userId: string,
feedUrl: string,
item: RssFeedItem,
@ -321,7 +255,8 @@ const createTask = async (
return createItemWithPreviewContent(userId, feedUrl, item)
}
return addFetchContentTask(userId, folder, item)
console.log(`adding fetch content task ${userId} ${item.link.trim()}`)
return addFetchContentTask(fetchContentTasks, userId, folder, item)
}
const fetchContentAndCreateItem = async (
@ -479,6 +414,7 @@ const getLink = (links: RssFeedItemLink[]): string | undefined => {
}
const processSubscription = async (
fetchContentTasks: Map<string, FetchContentTask>,
subscriptionId: string,
userId: string,
feedUrl: string,
@ -517,7 +453,7 @@ const processSubscription = async (
// use published or updated if isoDate is not available for atom feeds
const isoDate =
item.isoDate || item.published || item.updated || item.created
console.log('Processing feed item', item.links, item.isoDate, feed.feedUrl)
console.log('Processing feed item', item.links, item.isoDate, feedUrl)
if (!item.links || item.links.length === 0) {
console.log('Invalid feed item', item)
@ -530,7 +466,6 @@ const processSubscription = async (
continue
}
console.log('Fetching feed item', link)
const feedItem = {
...item,
isoDate,
@ -560,6 +495,7 @@ const processSubscription = async (
}
const created = await createTask(
fetchContentTasks,
userId,
feedUrl,
feedItem,
@ -589,6 +525,7 @@ const processSubscription = async (
// the feed has never been fetched, save at least the last valid item
const created = await createTask(
fetchContentTasks,
userId,
feedUrl,
lastValidItem,
@ -610,13 +547,11 @@ const processSubscription = async (
const nextScheduledAt = scheduledAt + updatePeriodInMs * updateFrequency
// update subscription lastFetchedAt
const updatedSubscription = await sendUpdateSubscriptionMutation(
userId,
subscriptionId,
lastItemFetchedAt,
updatedLastFetchedChecksum,
new Date(nextScheduledAt)
)
const updatedSubscription = await updateSubscription(userId, subscriptionId, {
lastFetchedAt: lastItemFetchedAt,
lastFetchedChecksum: updatedLastFetchedChecksum,
scheduledAt: new Date(nextScheduledAt),
})
console.log('Updated subscription', updatedSubscription)
}
@ -639,8 +574,9 @@ export const _refreshFeed = async (request: RefreshFeedRequest) => {
lastFetchedChecksums,
fetchContents,
folders,
refreshContext,
} = request
console.log('Processing feed', feedUrl)
console.log('Processing feed', feedUrl, { refreshContext: refreshContext })
const isBlocked = await isFeedBlocked(feedUrl)
if (isBlocked) {
@ -670,9 +606,11 @@ export const _refreshFeed = async (request: RefreshFeedRequest) => {
console.log('Fetched feed', feed.title, new Date())
const fetchContentTasks = new Map<string, FetchContentTask>() // url -> FetchContentTask
// process each subscription sequentially
for (let i = 0; i < subscriptionIds.length; i++) {
await processSubscription(
fetchContentTasks,
subscriptionIds[i],
userIds[i],
feedUrl,

122
packages/api/src/jobs/save_page.ts
View file

@ -3,6 +3,14 @@ import jwt from 'jsonwebtoken'
import { promisify } from 'util'
import { env } from '../env'
import { redisDataSource } from '../redis_data_source'
import { savePage } from '../services/save_page'
import { userRepository } from '../repository/user'
import { logger } from '../utils/logger'
import { Readability } from '@omnivore/readability'
import {
ArticleSavingRequestStatus,
CreateLabelInput,
} from '../generated/graphql'
const signToken = promisify(jwt.sign)
@ -18,7 +26,7 @@ interface Data {
url: string
articleSavingRequestId: string
state?: string
labels?: string[]
labels?: CreateLabelInput[]
source: string
folder: string
rssFeedUrl?: string
@ -65,7 +73,7 @@ interface FetchResult {
title: string
content?: string
contentType?: string
readabilityResult?: unknown
readabilityResult?: Readability.ParseResult
}
const isFetchResult = (obj: unknown): obj is FetchResult => {
@ -234,60 +242,6 @@ const sendCreateArticleMutation = async (userId: string, input: unknown) => {
}
}
const sendSavePageMutation = async (userId: string, input: unknown) => {
const data = JSON.stringify({
query: `mutation SavePage ($input: SavePageInput!){
savePage(input:$input){
... on SaveSuccess{
url
clientRequestId
}
... on SaveError{
errorCodes
}
}
}`,
variables: {
input,
},
})
const auth = await signToken({ uid: userId }, JWT_SECRET)
try {
const response = await axios.post<SavePageResponse>(
`${REST_BACKEND_ENDPOINT}/graphql`,
data,
{
headers: {
Cookie: `auth=${auth as string};`,
'Content-Type': 'application/json',
},
timeout: REQUEST_TIMEOUT,
}
)
if (
response.data.data.savePage.errorCodes &&
response.data.data.savePage.errorCodes.length > 0
) {
console.error(
'error while saving page',
response.data.data.savePage.errorCodes[0]
)
if (response.data.data.savePage.errorCodes[0] === 'UNAUTHORIZED') {
return { error: 'UNAUTHORIZED' }
}
return null
}
return response.data.data.savePage
} catch (error) {
console.error('error saving page', error)
return null
}
}
const sendImportStatusUpdate = async (
userId: string,
taskId: string,
@ -353,6 +307,7 @@ export const savePageJob = async (data: Data, attemptsMade: number) => {
try {
const url = encodeURI(data.url)
console.log(`savePageJob: ${userId} ${url}`)
// get the fetch result from cache
const { title, content, contentType, readabilityResult } =
@ -382,30 +337,45 @@ export const savePageJob = async (data: Data, attemptsMade: number) => {
return true
}
// for non-pdf content, we need to save the page
const apiResponse = await sendSavePageMutation(userId, {
url,
clientRequestId: articleSavingRequestId,
title,
originalContent: content,
parseResult: readabilityResult,
state,
labels,
rssFeedUrl,
savedAt,
publishedAt,
source,
folder,
})
if (!apiResponse) {
throw new Error('error while saving page')
if (!content) {
throw new Error(
'Invalid SavePage job, fetch result missing required data'
)
}
if ('error' in apiResponse && apiResponse.error === 'UNAUTHORIZED') {
console.log('user is deleted', userId)
return false
const user = await userRepository.findById(userId)
if (!user) {
logger.error('Unable to save job, user can not be found.', {
userId,
url,
})
throw new Error('Unable to save job, user can not be found.')
}
// for non-pdf content, we need to save the page
const result = await savePage(
{
url,
clientRequestId: articleSavingRequestId,
title,
originalContent: content,
parseResult: readabilityResult,
state: state ? (state as ArticleSavingRequestStatus) : undefined,
labels: labels,
rssFeedUrl,
savedAt: savedAt ? new Date(savedAt) : new Date(),
publishedAt: publishedAt ? new Date(publishedAt) : null,
source,
folder,
},
user
)
// if (result.__typename == 'SaveError') {
// logger.error('Error saving page', { userId, url, result })
// throw new Error('Error saving page')
// }
// if the readability result is not parsed, the import is failed
isImported = !!readabilityResult
isSaved = true

16
packages/api/src/queue-processor.ts
View file

@ -15,6 +15,22 @@ import { CustomTypeOrmLogger } from './utils/logger'
export const QUEUE_NAME = 'omnivore-backend-queue'
let backendQueue: Queue | undefined
export const getBackendQueue = async (): Promise<Queue | undefined> => {
if (backendQueue) {
await backendQueue.waitUntilReady()
return backendQueue
}
if (!redisDataSource.workerRedisClient) {
throw new Error('Can not create queues, redis is not initialized')
}
backendQueue = new Queue(QUEUE_NAME, {
connection: redisDataSource.workerRedisClient,
})
await backendQueue.waitUntilReady()
return backendQueue
}
const main = async () => {
console.log('[queue-processor]: starting queue processor')

2
packages/api/src/resolvers/api_key/index.ts
View file

@ -17,7 +17,7 @@ import { getRepository } from '../../repository'
import { findApiKeys } from '../../services/api_key'
import { analytics } from '../../utils/analytics'
import { generateApiKey, hashApiKey } from '../../utils/auth'
import { authorized } from '../../utils/helpers'
import { authorized } from '../../utils/gql-utils'
export const apiKeysResolver = authorized<ApiKeysSuccess, ApiKeysError>(
async (_, __, { log, uid }) => {

2
packages/api/src/resolvers/article/index.ts
View file

@ -93,7 +93,6 @@ import { traceAs } from '../../tracing'
import { analytics } from '../../utils/analytics'
import { isSiteBlockedForParse } from '../../utils/blocked'
import {
authorized,
cleanUrl,
errorHandler,
generateSlug,
@ -103,6 +102,7 @@ import {
titleForFilePath,
userDataToUser,
} from '../../utils/helpers'
import { authorized } from '../../utils/gql-utils'
import {
contentConverter,
getDistillerResult,

3
packages/api/src/resolvers/article_saving_request/index.ts
View file

@ -19,11 +19,12 @@ import {
} from '../../services/library_item'
import { analytics } from '../../utils/analytics'
import {
authorized,
cleanUrl,
isParsingTimeout,
libraryItemToArticleSavingRequest,
} from '../../utils/helpers'
import { authorized } from '../../utils/gql-utils'
import { isErrorWithCode } from '../user'
export const createArticleSavingRequestResolver = authorized<

2
packages/api/src/resolvers/features/index.ts
View file

@ -9,7 +9,7 @@ import {
optInFeature,
signFeatureToken,
} from '../../services/features'
import { authorized } from '../../utils/helpers'
import { authorized } from '../../utils/gql-utils'
export const optInFeatureResolver = authorized<
OptInFeatureSuccess,

2
packages/api/src/resolvers/filters/index.ts
View file

@ -25,7 +25,7 @@ import {
} from '../../generated/graphql'
import { authTrx } from '../../repository'
import { analytics } from '../../utils/analytics'
import { authorized } from '../../utils/helpers'
import { authorized } from '../../utils/gql-utils'
export const saveFilterResolver = authorized<
SaveFilterSuccess,

3
packages/api/src/resolvers/highlight/index.ts
View file

@ -34,7 +34,8 @@ import {
updateHighlight,
} from '../../services/highlights'
import { analytics } from '../../utils/analytics'
import { authorized, highlightDataToHighlight } from '../../utils/helpers'
import { highlightDataToHighlight } from '../../utils/helpers'
import { authorized } from '../../utils/gql-utils'
export const createHighlightResolver = authorized<
CreateHighlightSuccess,

2
packages/api/src/resolvers/importers/uploadImportFileResolver.ts
View file

@ -9,7 +9,7 @@ import {
} from '../../generated/graphql'
import { userRepository } from '../../repository/user'
import { analytics } from '../../utils/analytics'
import { authorized } from '../../utils/helpers'
import { authorized } from '../../utils/gql-utils'
import { logger } from '../../utils/logger'
import {
countOfFilesWithPrefix,

2
packages/api/src/resolvers/integrations/index.ts
View file

@ -37,7 +37,7 @@ import {
enqueueExportToIntegration,
enqueueImportFromIntegration,
} from '../../utils/createTask'
import { authorized } from '../../utils/helpers'
import { authorized } from '../../utils/gql-utils'
export const setIntegrationResolver = authorized<
SetIntegrationSuccess,

2
packages/api/src/resolvers/labels/index.ts
View file

@ -37,7 +37,7 @@ import {
updateLabel,
} from '../../services/labels'
import { analytics } from '../../utils/analytics'
import { authorized } from '../../utils/helpers'
import { authorized } from '../../utils/gql-utils'
export const labelsResolver = authorized<LabelsSuccess, LabelsError>(
async (_obj, _params, { authTrx, log, uid }) => {

2
packages/api/src/resolvers/links/index.ts
View file

@ -8,7 +8,7 @@ import {
} from '../../generated/graphql'
import { updateLibraryItem } from '../../services/library_item'
import { analytics } from '../../utils/analytics'
import { authorized } from '../../utils/helpers'
import { authorized } from '../../utils/gql-utils'
// export const updateLinkShareInfoResolver = authorized<
// UpdateLinkShareInfoSuccess,

2
packages/api/src/resolvers/newsletters/index.ts
View file

@ -30,7 +30,7 @@ import {
import { unsubscribeAll } from '../../services/subscriptions'
import { Merge } from '../../util'
import { analytics } from '../../utils/analytics'
import { authorized } from '../../utils/helpers'
import { authorized } from '../../utils/gql-utils'
export type CreateNewsletterEmailSuccessPartial = Merge<
CreateNewsletterEmailSuccess,

2
packages/api/src/resolvers/popular_reads/index.ts
View file

@ -5,7 +5,7 @@ import {
MutationAddPopularReadArgs,
} from '../../generated/graphql'
import { addPopularRead } from '../../services/popular_reads'
import { authorized } from '../../utils/helpers'
import { authorized } from '../../utils/gql-utils'
export const addPopularReadResolver = authorized<
AddPopularReadSuccess,
AddPopularReadError,

2
packages/api/src/resolvers/recent_emails/index.ts
View file

@ -13,7 +13,7 @@ import {
} from '../../generated/graphql'
import { updateReceivedEmail } from '../../services/received_emails'
import { saveNewsletter } from '../../services/save_newsletter_email'
import { authorized } from '../../utils/helpers'
import { authorized } from '../../utils/gql-utils'
import { generateUniqueUrl, parseEmailAddress } from '../../utils/parser'
import { sendEmail } from '../../utils/sendEmail'

2
packages/api/src/resolvers/recent_searches/index.ts
View file

@ -3,7 +3,7 @@ import {
RecentSearchesSuccess,
} from '../../generated/graphql'
import { getRecentSearches } from '../../services/search_history'
import { authorized } from '../../utils/helpers'
import { authorized } from '../../utils/gql-utils'
export const recentSearchesResolver = authorized<
RecentSearchesSuccess,

3
packages/api/src/resolvers/recommendations/index.ts
View file

@ -40,7 +40,8 @@ import {
import { findLibraryItemById } from '../../services/library_item'
import { analytics } from '../../utils/analytics'
import { enqueueRecommendation } from '../../utils/createTask'
import { authorized, userDataToUser } from '../../utils/helpers'
import { userDataToUser } from '../../utils/helpers'
import { authorized } from '../../utils/gql-utils'
export const createGroupResolver = authorized<
CreateGroupSuccess,

2
packages/api/src/resolvers/rules/index.ts
View file

@ -14,7 +14,7 @@ import {
SetRuleSuccess,
} from '../../generated/graphql'
import { deleteRule } from '../../services/rules'
import { authorized } from '../../utils/helpers'
import { authorized } from '../../utils/gql-utils'
export const setRuleResolver = authorized<
SetRuleSuccess,

2
packages/api/src/resolvers/save/index.ts
View file

@ -12,7 +12,7 @@ import { saveFile } from '../../services/save_file'
import { savePage } from '../../services/save_page'
import { saveUrl } from '../../services/save_url'
import { analytics } from '../../utils/analytics'
import { authorized } from '../../utils/helpers'
import { authorized } from '../../utils/gql-utils'
export const savePageResolver = authorized<
SaveSuccess,

2
packages/api/src/resolvers/send_install_instructions/index.ts
View file

@ -5,7 +5,7 @@ import {
SendInstallInstructionsSuccess,
} from '../../generated/graphql'
import { userRepository } from '../../repository/user'
import { authorized } from '../../utils/helpers'
import { authorized } from '../../utils/gql-utils'
import { sendEmail } from '../../utils/sendEmail'
const INSTALL_INSTRUCTIONS_EMAIL_TEMPLATE_ID =

37
packages/api/src/resolvers/subscriptions/index.ts
View file

@ -44,12 +44,10 @@ import { unsubscribe } from '../../services/subscriptions'
import { Merge } from '../../util'
import { analytics } from '../../utils/analytics'
import { enqueueRssFeedFetch } from '../../utils/createTask'
import {
authorized,
getAbsoluteUrl,
keysToCamelCase,
} from '../../utils/helpers'
import { getAbsoluteUrl, keysToCamelCase } from '../../utils/helpers'
import { authorized } from '../../utils/gql-utils'
import { parseFeed, parseOpml, RSS_PARSER_CONFIG } from '../../utils/parser'
import { updateSubscription } from '../../services/update_subscription'
type PartialSubscription = Omit<Subscription, 'newsletterEmail'>
@ -332,34 +330,7 @@ export const updateSubscriptionResolver = authorized<
},
})
const updatedSubscription = await authTrx(async (t) => {
const repo = t.getRepository(Subscription)
// update subscription
await t.getRepository(Subscription).save({
id: input.id,
name: input.name || undefined,
description: input.description || undefined,
lastFetchedAt: input.lastFetchedAt
? new Date(input.lastFetchedAt)
: undefined,
lastFetchedChecksum: input.lastFetchedChecksum || undefined,
status: input.status || undefined,
scheduledAt: input.scheduledAt
? new Date(input.scheduledAt)
: undefined,
autoAddToLibrary: input.autoAddToLibrary ?? undefined,
isPrivate: input.isPrivate ?? undefined,
fetchContent: input.fetchContent ?? undefined,
folder: input.folder ?? undefined,
})
return repo.findOneByOrFail({
id: input.id,
user: { id: uid },
})
})
const updatedSubscription = await updateSubscription(uid, input.id, input)
return {
subscription: updatedSubscription,
}

3
packages/api/src/resolvers/update/index.ts
View file

@ -5,7 +5,8 @@ import {
UpdatePageSuccess,
} from '../../generated/graphql'
import { updateLibraryItem } from '../../services/library_item'
import { authorized, libraryItemToArticle } from '../../utils/helpers'
import { libraryItemToArticle } from '../../utils/helpers'
import { authorized } from '../../utils/gql-utils'
export const updatePageResolver = authorized<
UpdatePageSuccess,

4
packages/api/src/resolvers/upload_files/index.ts
View file

@ -19,7 +19,9 @@ import {
updateLibraryItem,
} from '../../services/library_item'
import { analytics } from '../../utils/analytics'
import { authorized, generateSlug } from '../../utils/helpers'
import { generateSlug } from '../../utils/helpers'
import { authorized } from '../../utils/gql-utils'
import {
contentReaderForLibraryItem,
generateUploadFilePathName,

3
packages/api/src/resolvers/user/index.ts
View file

@ -43,9 +43,10 @@ import { userRepository } from '../../repository/user'
import { createUser } from '../../services/create_user'
import { sendVerificationEmail } from '../../services/send_emails'
import { softDeleteUser } from '../../services/user'
import { authorized, userDataToUser } from '../../utils/helpers'
import { userDataToUser } from '../../utils/helpers'
import { validateUsername } from '../../utils/usernamePolicy'
import { WithDataSourcesContext } from '../types'
import { authorized } from '../../utils/gql-utils'
export const updateUserResolver = authorized<
UpdateUserSuccess,

2
packages/api/src/resolvers/user_device_tokens/index.ts
View file

@ -20,7 +20,7 @@ import {
findDeviceTokensByUserId,
} from '../../services/user_device_tokens'
import { analytics } from '../../utils/analytics'
import { authorized } from '../../utils/helpers'
import { authorized } from '../../utils/gql-utils'
const PG_UNIQUE_CONSTRAINT_VIOLATION = '23505'

2
packages/api/src/resolvers/user_personalization/index.ts
View file

@ -8,7 +8,7 @@ import {
SetUserPersonalizationSuccess,
SortOrder,
} from '../../generated/graphql'
import { authorized } from '../../utils/helpers'
import { authorized } from '../../utils/gql-utils'
export const setUserPersonalizationResolver = authorized<
SetUserPersonalizationSuccess,

2
packages/api/src/resolvers/webhooks/index.ts
View file

@ -22,7 +22,7 @@ import {
import { authTrx } from '../../repository'
import { deleteWebhook } from '../../services/webhook'
import { analytics } from '../../utils/analytics'
import { authorized } from '../../utils/helpers'
import { authorized } from '../../utils/gql-utils'
export const webhooksResolver = authorized<WebhooksSuccess, WebhooksError>(
async (_obj, _params, { uid, log }) => {

3
packages/api/src/services/save_page.ts
View file

@ -7,7 +7,6 @@ import { User } from '../entity/user'
import { homePageURL } from '../env'
import {
ArticleSavingRequestStatus,
Maybe,
PreparedDocumentInput,
SaveErrorCode,
SavePageInput,
@ -43,7 +42,7 @@ const ALREADY_PARSED_SOURCES = [
'pocket',
]
const createSlug = (url: string, title?: Maybe<string> | undefined) => {
const createSlug = (url: string, title?: string | null | undefined) => {
const { pathname } = new URL(url)
const croppedPathname = decodeURIComponent(
pathname

63
packages/api/src/services/update_subscription.ts Normal file
View file

@ -0,0 +1,63 @@
import { Subscription } from '../entity/subscription'
import {
SubscriptionStatus,
UpdateSubscriptionInput,
} from '../generated/graphql'
import { getRepository } from '../repository'
const ensureOwns = async (userId: string, subscriptionId: string) => {
const repo = getRepository(Subscription)
const existing = await repo.findOneByOrFail({
id: subscriptionId,
user: { id: userId },
})
if (!existing) {
throw new Error('Can not find subscription being updated.')
}
}
type UpdateSubscriptionData = {
autoAddToLibrary?: boolean | null
description?: string | null
fetchContent?: boolean | null
folder?: string | null
isPrivate?: boolean | null
lastFetchedAt?: Date | null
lastFetchedChecksum?: string | null
name?: string | null
scheduledAt?: Date | null
status?: SubscriptionStatus | null
}
export const updateSubscription = async (
userId: string,
subscriptionId: string,
newData: UpdateSubscriptionData
): Promise<Subscription> => {
await ensureOwns(userId, subscriptionId)
const repo = getRepository(Subscription)
await repo.save({
id: subscriptionId,
name: newData.name || undefined,
description: newData.description || undefined,
lastFetchedAt: newData.lastFetchedAt
? new Date(newData.lastFetchedAt)
: undefined,
lastFetchedChecksum: newData.lastFetchedChecksum || undefined,
status: newData.status || undefined,
scheduledAt: newData.scheduledAt
? new Date(newData.scheduledAt)
: undefined,
autoAddToLibrary: newData.autoAddToLibrary ?? undefined,
isPrivate: newData.isPrivate ?? undefined,
fetchContent: newData.fetchContent ?? undefined,
folder: newData.folder ?? undefined,
})
return await getRepository(Subscription).findOneByOrFail({
id: subscriptionId,
user: { id: userId },
})
}

3
packages/api/src/util.ts
View file

@ -68,7 +68,6 @@ export interface BackendEnv {
textToSpeechTaskHandlerUrl: string
recommendationTaskHandlerUrl: string
thumbnailTaskHandlerUrl: string
rssFeedTaskHandlerUrl: string
integrationExporterUrl: string
integrationImporterUrl: string
importerMetricsUrl: string
@ -149,7 +148,6 @@ const nullableEnvVars = [
'RECOMMENDATION_TASK_HANDLER_URL',
'POCKET_CONSUMER_KEY',
'THUMBNAIL_TASK_HANDLER_URL',
'RSS_FEED_TASK_HANDLER_URL',
'SENDGRID_VERIFICATION_TEMPLATE_ID',
'REMINDER_TASK_HANDLER_URL',
'TRUST_PROXY',
@ -247,7 +245,6 @@ export function getEnv(): BackendEnv {
textToSpeechTaskHandlerUrl: parse('TEXT_TO_SPEECH_TASK_HANDLER_URL'),
recommendationTaskHandlerUrl: parse('RECOMMENDATION_TASK_HANDLER_URL'),
thumbnailTaskHandlerUrl: parse('THUMBNAIL_TASK_HANDLER_URL'),
rssFeedTaskHandlerUrl: parse('RSS_FEED_TASK_HANDLER_URL'),
integrationExporterUrl: parse('INTEGRATION_EXPORTER_URL'),
integrationImporterUrl: parse('INTEGRATION_IMPORTER_URL'),
importerMetricsUrl: parse('IMPORTER_METRICS_COLLECTOR_URL'),

41
packages/api/src/utils/createTask.ts
View file

@ -22,6 +22,7 @@ import View = google.cloud.tasks.v2.Task.View
import { stringToHash } from './helpers'
import { queueRSSRefreshFeedJob } from '../jobs/rss/refreshAllFeeds'
import { redisDataSource } from '../redis_data_source'
import { v4 as uuid } from 'uuid'
// Instantiates a client.
const client = new CloudTasksClient()
@ -639,8 +640,12 @@ export interface RssSubscriptionGroup {
export const enqueueRssFeedFetch = async (
subscriptionGroup: RssSubscriptionGroup
): Promise<string> => {
const { GOOGLE_CLOUD_PROJECT, PUBSUB_VERIFICATION_TOKEN } = process.env
const payload = {
refreshContext: {
type: 'user-added',
refreshID: uuid(),
startedAt: new Date().toISOString(),
},
subscriptionIds: subscriptionGroup.subscriptionIds,
feedUrl: subscriptionGroup.url,
lastFetchedTimestamps: subscriptionGroup.fetchedDates.map(
@ -670,40 +675,6 @@ export const enqueueRssFeedFetch = async (
} else {
throw 'unable to queue rss-refresh-feed-job, redis is not configured'
}
// // If there is no Google Cloud Project Id exposed, it means that we are in local environment
// if (env.dev.isLocal || !GOOGLE_CLOUD_PROJECT) {
// if (env.queue.rssFeedTaskHandlerUrl) {
// // Calling the handler function directly.
// setTimeout(() => {
// axios
// .post(
// `${env.queue.rssFeedTaskHandlerUrl}?token=${PUBSUB_VERIFICATION_TOKEN}`,
// payload
// )
// .catch((error) => {
// logError(error)
// })
// }, 0)
// }
// return nanoid()
// }
// const createdTasks = await createHttpTaskWithToken({
// project: GOOGLE_CLOUD_PROJECT,
// queue: 'omnivore-rss-queue',
// payload,
// taskHandlerUrl: `${env.queue.rssFeedTaskHandlerUrl}?token=${PUBSUB_VERIFICATION_TOKEN}`,
// })
// if (!createdTasks || !createdTasks[0].name) {
// logger.error(`Unable to get the name of the task`, {
// payload,
// createdTasks,
// })
// throw new CreateTaskError(`Unable to get the name of the task`)
// }
//return createdTasks[0].name
}
export default createHttpTaskWithToken

26
packages/api/src/utils/gql-utils.ts Normal file
View file

@ -0,0 +1,26 @@
import { ResolverFn } from '../generated/graphql'
import { Claims, WithDataSourcesContext } from '../resolvers/types'
export function authorized<
TSuccess,
TError extends { errorCodes: string[] },
/* eslint-disable @typescript-eslint/no-explicit-any */
TArgs = any,
TParent = any
/* eslint-enable @typescript-eslint/no-explicit-any */
>(
resolver: ResolverFn<
TSuccess | TError,
TParent,
WithDataSourcesContext & { claims: Claims },
TArgs
>
): ResolverFn<TSuccess | TError, TParent, WithDataSourcesContext, TArgs> {
return (parent, args, ctx, info) => {
const { claims } = ctx
if (claims?.uid) {
return resolver(parent, args, { ...ctx, claims, uid: claims.uid }, info)
}
return { errorCodes: ['UNAUTHORIZED'] } as TError
}
}

24
packages/api/src/utils/helpers.ts
View file

@ -78,30 +78,6 @@ export const stringToHash = (str: string, convertToUUID = false): string => {
).toLowerCase()
}
export function authorized<
TSuccess,
TError extends { errorCodes: string[] },
/* eslint-disable @typescript-eslint/no-explicit-any */
TArgs = any,
TParent = any
/* eslint-enable @typescript-eslint/no-explicit-any */
>(
resolver: ResolverFn<
TSuccess | TError,
TParent,
WithDataSourcesContext & { claims: Claims },
TArgs
>
): ResolverFn<TSuccess | TError, TParent, WithDataSourcesContext, TArgs> {
return (parent, args, ctx, info) => {
const { claims } = ctx
if (claims?.uid) {
return resolver(parent, args, { ...ctx, claims, uid: claims.uid }, info)
}
return { errorCodes: ['UNAUTHORIZED'] } as TError
}
}
export const findDelimiter = (
text: string,
delimiters = ['\t', ',', ':', ';'],

89
packages/api/test/resolvers/article.test.ts
View file

@ -17,7 +17,7 @@ import {
PageType,
SyncUpdatedItemEdge,
UpdateReason,
UploadFileStatus
UploadFileStatus,
} from '../../src/generated/graphql'
import { getRepository } from '../../src/repository'
import { createGroup, deleteGroup } from '../../src/services/groups'
@ -25,7 +25,7 @@ import { createHighlight } from '../../src/services/highlights'
import {
createLabel,
deleteLabels,
saveLabelsInLibraryItem
saveLabelsInLibraryItem,
} from '../../src/services/labels'
import {
createLibraryItem,
@ -36,7 +36,7 @@ import {
deleteLibraryItemsByUserId,
findLibraryItemById,
findLibraryItemByUrl,
updateLibraryItem
updateLibraryItem,
} from '../../src/services/library_item'
import { deleteUser } from '../../src/services/user'
import * as createTask from '../../src/utils/createTask'
@ -570,23 +570,37 @@ describe('Article API', () => {
).expect(200)
// Save a link, then archive it
let allLinks = await graphqlRequest(searchQuery('in:inbox'), authToken).expect(
200
)
let allLinks = await graphqlRequest(
searchQuery('in:inbox'),
authToken
).expect(200)
const justSavedId = allLinks.body.data.search.edges[0].node.id
await archiveLink(authToken, justSavedId)
// test the negative case, ensuring the archive link wasn't returned
allLinks = await graphqlRequest(searchQuery('in:inbox'), authToken).expect(200)
allLinks = await graphqlRequest(
searchQuery('in:inbox'),
authToken
).expect(200)
expect(allLinks.body.data.search.edges[0]?.node?.url).to.not.eq(url)
// Now save the link again, and ensure it is returned
await graphqlRequest(
savePageQuery(url, title, originalContent, null, null, generateFakeUuid()),
savePageQuery(
url,
title,
originalContent,
null,
null,
generateFakeUuid()
),
authToken
).expect(200)
allLinks = await graphqlRequest(searchQuery('in:inbox'), authToken).expect(200)
allLinks = await graphqlRequest(
searchQuery('in:inbox'),
authToken
).expect(200)
expect(allLinks.body.data.search.edges[0].node.id).to.eq(justSavedId)
expect(allLinks.body.data.search.edges[0].node.url).to.eq(url)
})
@ -610,6 +624,7 @@ describe('Article API', () => {
).expect(200)
const savedItem = await findLibraryItemByUrl(url, user.id)
console.log('savedItem: ', savedItem)
expect(savedItem?.archivedAt).to.not.be.null
expect(savedItem?.labels?.map((l) => l.name)).to.eql(labels)
})
@ -778,15 +793,20 @@ describe('Article API', () => {
context('when force is true', () => {
before(async () => {
itemId = (await createLibraryItem({
user: { id: user.id },
originalUrl: 'https://blog.omnivore.app/setBookmarkArticle',
slug: 'test-with-omnivore',
readableContent: '<p>test</p>',
title: 'test title',
readingProgressBottomPercent: 100,
readingProgressTopPercent: 80,
}, user.id)).id
itemId = (
await createLibraryItem(
{
user: { id: user.id },
originalUrl: 'https://blog.omnivore.app/setBookmarkArticle',
slug: 'test-with-omnivore',
readableContent: '<p>test</p>',
title: 'test title',
readingProgressBottomPercent: 100,
readingProgressTopPercent: 80,
},
user.id
)
).id
})
after(async () => {
@ -2052,20 +2072,23 @@ describe('Article API', () => {
)
})
context('when since is -1000000000-01-01T00:00:00Z from android app', () => {
before(() => {
since = '-1000000000-01-01T00:00:00Z'
})
context(
'when since is -1000000000-01-01T00:00:00Z from android app',
() => {
before(() => {
since = '-1000000000-01-01T00:00:00Z'
})
it('returns all', async () => {
const res = await graphqlRequest(
updatesSinceQuery(since),
authToken
).expect(200)
it('returns all', async () => {
const res = await graphqlRequest(
updatesSinceQuery(since),
authToken
).expect(200)
expect(res.body.data.updatesSince.edges.length).to.eql(5)
})
})
expect(res.body.data.updatesSince.edges.length).to.eql(5)
})
}
)
context('returns highlights', () => {
let highlight: Highlight
@ -2092,9 +2115,9 @@ describe('Article API', () => {
expect(res.body.data.updatesSince.edges[0].node.highlights[0].id).to.eq(
highlight.id
)
expect(res.body.data.updatesSince.edges[0].node.highlights[0].type).to.eq(
HighlightType.Highlight
)
expect(
res.body.data.updatesSince.edges[0].node.highlights[0].type
).to.eq(HighlightType.Highlight)
})
})
})

1
packages/content-fetch/src/job.ts
View file

@ -57,6 +57,7 @@ export const queueSavePageJob = async (savePageJobs: savePageJob[]) => {
data: job.data,
opts: getOpts(job),
}))
console.log('queue save page jobs:', { jobs })
return queue.addBulk(jobs)
}