From fb576a58a01e61d07ad0cb4e84c798e1ad90cafd Mon Sep 17 00:00:00 2001 From: Hongbo Wu Date: Thu, 22 Feb 2024 15:20:24 +0800 Subject: [PATCH] initial export --- .../src/jobs/integration/export_all_items.ts | 69 +++++++++++++++++++ .../src/jobs/{ => integration}/export_item.ts | 32 +++++---- packages/api/src/pubsub.ts | 8 +-- packages/api/src/queue-processor.ts | 11 ++- .../api/src/resolvers/integrations/index.ts | 15 +--- packages/api/src/routers/svc/integrations.ts | 4 +- packages/api/src/services/library_item.ts | 15 ++++ packages/api/src/utils/createTask.ts | 62 +++++------------ 8 files changed, 137 insertions(+), 79 deletions(-) create mode 100644 packages/api/src/jobs/integration/export_all_items.ts rename packages/api/src/jobs/{ => integration}/export_item.ts (66%) diff --git a/packages/api/src/jobs/integration/export_all_items.ts b/packages/api/src/jobs/integration/export_all_items.ts new file mode 100644 index 000000000..fb29b20fb --- /dev/null +++ b/packages/api/src/jobs/integration/export_all_items.ts @@ -0,0 +1,69 @@ +import { IntegrationType } from '../../entity/integration' +import { findIntegration } from '../../services/integrations' +import { searchLibraryItems } from '../../services/library_item' +import { findActiveUser } from '../../services/user' +import { enqueueExportItem } from '../../utils/createTask' +import { logger } from '../../utils/logger' + +export interface ExportAllItemsJobData { + userId: string + integrationId: string +} + +export const EXPORT_ALL_ITEMS_JOB_NAME = 'export-all-items' + +export const exportAllItems = async (jobData: ExportAllItemsJobData) => { + const { userId, integrationId } = jobData + const user = await findActiveUser(userId) + if (!user) { + logger.error('user not found', { + userId, + }) + return + } + + const integration = await findIntegration( + { + id: integrationId, + enabled: true, + type: IntegrationType.Export, + }, + userId + ) + + if (!integration) { + logger.error('integration not found', { + userId, + integrationId, + }) + return + } + + // get paginated items from the database + const first = 50 + let after = 0 + for (;;) { + console.log('searching for items...', { + userId, + first, + after, + }) + const searchResult = await searchLibraryItems( + { from: after, size: first }, + userId + ) + const libraryItems = searchResult.libraryItems + const size = libraryItems.length + if (size === 0) { + break + } + + await enqueueExportItem({ + userId, + libraryItemIds: libraryItems.map((item) => item.id), + integrationId, + }) + + after += size + } +} diff --git a/packages/api/src/jobs/export_item.ts b/packages/api/src/jobs/integration/export_item.ts similarity index 66% rename from packages/api/src/jobs/export_item.ts rename to packages/api/src/jobs/integration/export_item.ts index b2e13ea35..896ca45b4 100644 --- a/packages/api/src/jobs/export_item.ts +++ b/packages/api/src/jobs/integration/export_item.ts @@ -1,31 +1,33 @@ -import { IntegrationType } from '../entity/integration' +import { IntegrationType } from '../../entity/integration' import { findIntegrations, getIntegrationClient, updateIntegration, -} from '../services/integrations' -import { findLibraryItemById } from '../services/library_item' -import { logger } from '../utils/logger' +} from '../../services/integrations' +import { findLibraryItemsByIds } from '../../services/library_item' +import { logger } from '../../utils/logger' export interface ExportItemJobData { userId: string - libraryItemId: string + libraryItemIds: string[] + integrationId?: string } export const EXPORT_ITEM_JOB_NAME = 'export-item' export const exportItem = async (jobData: ExportItemJobData) => { - const { libraryItemId, userId } = jobData - const libraryItem = await findLibraryItemById(libraryItemId, userId) - if (!libraryItem) { - logger.error('library item not found', { + const { libraryItemIds, userId, integrationId } = jobData + const libraryItems = await findLibraryItemsByIds(libraryItemIds, userId) + if (libraryItems.length === 0) { + logger.error('library items not found', { userId, - libraryItemId, + libraryItemIds, }) return } const integrations = await findIntegrations(userId, { + id: integrationId, enabled: true, type: IntegrationType.Export, }) @@ -38,7 +40,7 @@ export const exportItem = async (jobData: ExportItemJobData) => { integrations.map(async (integration) => { const logObject = { userId, - libraryItemId, + libraryItemIds, integrationId: integration.id, } logger.info('exporting item...', logObject) @@ -46,23 +48,23 @@ export const exportItem = async (jobData: ExportItemJobData) => { try { const client = getIntegrationClient(integration.name) - const synced = await client.export(integration.token, [libraryItem]) + const synced = await client.export(integration.token, libraryItems) if (!synced) { logger.error('failed to export item', logObject) return Promise.resolve(false) } - const lastItemUpdatedAt = libraryItem.updatedAt + const syncedAt = new Date() logger.info('updating integration...', { ...logObject, - syncedAt: lastItemUpdatedAt, + syncedAt, }) // update integration syncedAt if successful const updated = await updateIntegration( integration.id, { - syncedAt: lastItemUpdatedAt, + syncedAt, }, userId ) diff --git a/packages/api/src/pubsub.ts b/packages/api/src/pubsub.ts index 723fb7947..5ca9d2693 100644 --- a/packages/api/src/pubsub.ts +++ b/packages/api/src/pubsub.ts @@ -64,10 +64,10 @@ export const createPubSubClient = (): PubsubClient => { libraryItemId, }) } - + // queue export item job await enqueueExportItem({ userId, - libraryItemId, + libraryItemIds: [libraryItemId], }) const cleanData = deepDelete( @@ -102,10 +102,10 @@ export const createPubSubClient = (): PubsubClient => { libraryItemId, }) } - + // queue export item job await enqueueExportItem({ userId, - libraryItemId, + libraryItemIds: [libraryItemId], }) const cleanData = deepDelete( diff --git a/packages/api/src/queue-processor.ts b/packages/api/src/queue-processor.ts index f1b5908f9..cbd06787b 100644 --- a/packages/api/src/queue-processor.ts +++ b/packages/api/src/queue-processor.ts @@ -15,8 +15,15 @@ import { appDataSource } from './data_source' import { env } from './env' import { bulkAction, BULK_ACTION_JOB_NAME } from './jobs/bulk_action' import { callWebhook, CALL_WEBHOOK_JOB_NAME } from './jobs/call_webhook' -import { exportItem, EXPORT_ITEM_JOB_NAME } from './jobs/export_item' import { findThumbnail, THUMBNAIL_JOB } from './jobs/find_thumbnail' +import { + exportAllItems, + EXPORT_ALL_ITEMS_JOB_NAME, +} from './jobs/integration/export_all_items' +import { + exportItem, + EXPORT_ITEM_JOB_NAME, +} from './jobs/integration/export_item' import { refreshAllFeeds } from './jobs/rss/refreshAllFeeds' import { refreshFeed } from './jobs/rss/refreshFeed' import { savePageJob } from './jobs/save_page' @@ -106,6 +113,8 @@ export const createWorker = (connection: ConnectionOptions) => return callWebhook(job.data) case EXPORT_ITEM_JOB_NAME: return exportItem(job.data) + case EXPORT_ALL_ITEMS_JOB_NAME: + return exportAllItems(job.data) } }, { diff --git a/packages/api/src/resolvers/integrations/index.ts b/packages/api/src/resolvers/integrations/index.ts index 0a849c5f0..edab2d636 100644 --- a/packages/api/src/resolvers/integrations/index.ts +++ b/packages/api/src/resolvers/integrations/index.ts @@ -34,7 +34,7 @@ import { import { analytics } from '../../utils/analytics' import { deleteTask, - enqueueExportToIntegration, + enqueueExportAllItems, enqueueImportFromIntegration, } from '../../utils/createTask' import { authorized } from '../../utils/gql-utils' @@ -98,17 +98,8 @@ export const setIntegrationResolver = authorized< } // create a task to sync all the pages if new integration or enable integration (export type) - const taskName = await enqueueExportToIntegration( - integration.id, - integration.name, - 0, - authToken - ) - log.info('enqueued task', taskName) - - // update task name in integration - await updateIntegration(integration.id, { taskName }, uid) - integration.taskName = taskName + const job = await enqueueExportAllItems(integration.id, uid) + log.info('enqueued job', job) } else if (integrationToSave.taskName) { // delete the task if disable integration and task exists const result = await deleteTask(integrationToSave.taskName) diff --git a/packages/api/src/routers/svc/integrations.ts b/packages/api/src/routers/svc/integrations.ts index 68db6b846..aab6331f9 100644 --- a/packages/api/src/routers/svc/integrations.ts +++ b/packages/api/src/routers/svc/integrations.ts @@ -5,7 +5,7 @@ import express from 'express' import { Integration, IntegrationType } from '../../entity/integration' import { readPushSubscription } from '../../pubsub' import { getRepository } from '../../repository' -import { enqueueExportToIntegration } from '../../utils/createTask' +import { enqueueExportAllItems } from '../../utils/createTask' import { logger } from '../../utils/logger' import { createIntegrationToken } from '../auth/jwt_helpers' @@ -51,7 +51,7 @@ export function integrationsServiceRouter() { } const syncAt = integration.syncedAt?.getTime() || 0 - return enqueueExportToIntegration( + return enqueueExportAllItems( integration.id, integration.name, syncAt, diff --git a/packages/api/src/services/library_item.ts b/packages/api/src/services/library_item.ts index 38658be3c..4cb581784 100644 --- a/packages/api/src/services/library_item.ts +++ b/packages/api/src/services/library_item.ts @@ -652,6 +652,21 @@ export const searchLibraryItems = async ( ) } +export const findLibraryItemsByIds = async (ids: string[], userId: string) => { + return authTrx( + async (tx) => + tx + .createQueryBuilder(LibraryItem, 'library_item') + .leftJoinAndSelect('library_item.labels', 'labels') + .leftJoinAndSelect('library_item.highlights', 'highlights') + .leftJoinAndSelect('highlights.user', 'user') + .where('library_item.id IN (:...ids)', { ids }) + .getMany(), + undefined, + userId + ) +} + export const findLibraryItemById = async ( id: string, userId: string diff --git a/packages/api/src/utils/createTask.ts b/packages/api/src/utils/createTask.ts index 20967d05d..50ef58f45 100644 --- a/packages/api/src/utils/createTask.ts +++ b/packages/api/src/utils/createTask.ts @@ -16,8 +16,12 @@ import { } from '../generated/graphql' import { BulkActionData, BULK_ACTION_JOB_NAME } from '../jobs/bulk_action' import { CallWebhookJobData, CALL_WEBHOOK_JOB_NAME } from '../jobs/call_webhook' -import { ExportItemJobData, EXPORT_ITEM_JOB_NAME } from '../jobs/export_item' import { THUMBNAIL_JOB } from '../jobs/find_thumbnail' +import { EXPORT_ALL_ITEMS_JOB_NAME } from '../jobs/integration/export_all_items' +import { + ExportItemJobData, + EXPORT_ITEM_JOB_NAME, +} from '../jobs/integration/export_item' import { queueRSSRefreshFeedJob, REFRESH_ALL_FEEDS_JOB_NAME, @@ -67,6 +71,7 @@ export const getJobPriority = (jobName: string): number => { return 10 case `${REFRESH_FEED_JOB_NAME}_low`: return 50 + case EXPORT_ALL_ITEMS_JOB_NAME: case REFRESH_ALL_FEEDS_JOB_NAME: case THUMBNAIL_JOB: return 100 @@ -576,55 +581,22 @@ export const enqueueImportFromIntegration = async ( return createdTasks[0].name } -export const enqueueExportToIntegration = async ( +export const enqueueExportAllItems = async ( integrationId: string, - integrationName: string, - syncAt: number, // unix timestamp in milliseconds - authToken: string -): Promise => { - const { GOOGLE_CLOUD_PROJECT } = process.env + userId: string +) => { + const queue = await getBackendQueue() + if (!queue) { + return undefined + } const payload = { + userId, integrationId, - integrationName, - syncAt, } - - const headers = { - [OmnivoreAuthorizationHeader]: authToken, - } - // 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.integrationExporterUrl) { - // Calling the handler function directly. - setTimeout(() => { - axios - .post(env.queue.integrationExporterUrl, payload, { - headers, - }) - .catch((error) => { - logError(error) - }) - }, 0) - } - return nanoid() - } - - const createdTasks = await createHttpTaskWithToken({ - project: GOOGLE_CLOUD_PROJECT, - payload, - taskHandlerUrl: env.queue.integrationExporterUrl, - priority: 'low', - requestHeaders: headers, + return queue.add(EXPORT_ALL_ITEMS_JOB_NAME, payload, { + priority: getJobPriority(EXPORT_ALL_ITEMS_JOB_NAME), + attempts: 1, }) - - 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 const enqueueThumbnailJob = async (