From ad2e8ee002d0f31e325714cc6d54f2c7a71d1162 Mon Sep 17 00:00:00 2001 From: Hongbo Wu Date: Fri, 22 Mar 2024 11:47:30 +0800 Subject: [PATCH] trigger export in rules --- packages/api/src/entity/rule.ts | 1 + packages/api/src/jobs/trigger_rule.ts | 69 +++++++++++ packages/api/src/pubsub.ts | 14 +-- packages/api/src/services/highlights.ts | 28 ++++- .../src/services/integrations/integration.ts | 5 +- .../api/src/services/integrations/notion.ts | 114 ++++++++++-------- .../api/src/services/integrations/readwise.ts | 14 +-- packages/api/src/services/labels.ts | 13 +- 8 files changed, 182 insertions(+), 76 deletions(-) diff --git a/packages/api/src/entity/rule.ts b/packages/api/src/entity/rule.ts index 9ef6d797e..6dab77e60 100644 --- a/packages/api/src/entity/rule.ts +++ b/packages/api/src/entity/rule.ts @@ -16,6 +16,7 @@ export enum RuleActionType { MarkAsRead = 'MARK_AS_READ', SendNotification = 'SEND_NOTIFICATION', Webhook = 'WEBHOOK', + Export = 'EXPORT', } export enum RuleEventType { diff --git a/packages/api/src/jobs/trigger_rule.ts b/packages/api/src/jobs/trigger_rule.ts index 68d30e06b..550f9cacf 100644 --- a/packages/api/src/jobs/trigger_rule.ts +++ b/packages/api/src/jobs/trigger_rule.ts @@ -1,8 +1,14 @@ import { LiqeQuery } from '@omnivore/liqe' import axios from 'axios' import { ReadingProgressDataSource } from '../datasources/reading_progress_data_source' +import { IntegrationType } from '../entity/integration' import { LibraryItem, LibraryItemState } from '../entity/library_item' import { Rule, RuleAction, RuleActionType, RuleEventType } from '../entity/rule' +import { + findIntegrations, + getIntegrationClient, + updateIntegration, +} from '../services/integrations' import { addLabelsToLibraryItem } from '../services/labels' import { filterItemEvents, @@ -107,6 +113,67 @@ const sendToWebhook = async (obj: RuleActionObj) => { }) } +const exportItem = async (obj: RuleActionObj) => { + const userId = obj.userId + const integrations = await findIntegrations(userId, { + enabled: true, + type: IntegrationType.Export, + }) + + if (integrations.length <= 0) { + return + } + + await Promise.all( + integrations.map(async (integration) => { + try { + const logObject = { + userId, + integrationId: integration.id, + } + logger.info('exporting item...', logObject) + + const client = getIntegrationClient( + integration.name, + integration.token, + integration + ) + + const synced = await client.export([obj.data]) + if (!synced) { + logger.error('failed to export item', logObject) + return false + } + + const syncedAt = new Date() + logger.info('updating integration...', { + ...logObject, + syncedAt, + }) + + // update integration syncedAt if successful + const updated = await updateIntegration( + integration.id, + { + syncedAt, + }, + userId + ) + logger.info('integration updated', { + ...logObject, + updated, + }) + } catch (error) { + logger.error('failed to export item', { + userId, + integrationId: integration.id, + error, + }) + } + }) + ) +} + const getRuleAction = ( actionType: RuleActionType ): RuleActionFunc | undefined => { @@ -123,6 +190,8 @@ const getRuleAction = ( return sendNotification case RuleActionType.Webhook: return sendToWebhook + case RuleActionType.Export: + return exportItem default: logger.error('Unknown rule action type', actionType) return undefined diff --git a/packages/api/src/pubsub.ts b/packages/api/src/pubsub.ts index 5e761fe1b..27babb13b 100644 --- a/packages/api/src/pubsub.ts +++ b/packages/api/src/pubsub.ts @@ -4,7 +4,6 @@ import { RuleEventType } from './entity/rule' import { env } from './env' import { ReportType } from './generated/graphql' import { - enqueueExportItem, enqueueProcessYouTubeVideo, enqueueTriggerRuleJob, } from './utils/createTask' @@ -58,11 +57,6 @@ export const createPubSubClient = (): PubsubClient => { data, userId, }) - // queue export item job - await enqueueExportItem({ - userId, - libraryItemIds: [data.id], - }) if (type === EntityType.ITEM) { // if (await findGrantedFeatureByName(FeatureName.AISummaries, userId)) { @@ -92,15 +86,9 @@ export const createPubSubClient = (): PubsubClient => { // queue trigger rule job await enqueueTriggerRuleJob({ userId, - ruleEventType: RuleEventType.PageUpdated, + ruleEventType: `${type.toUpperCase()}_UPDATED` as RuleEventType, data, }) - - // queue export item job - await enqueueExportItem({ - userId, - libraryItemIds: [data.id], - }) }, entityDeleted: async ( type: EntityType, diff --git a/packages/api/src/services/highlights.ts b/packages/api/src/services/highlights.ts index 43d710d36..73a1bcb35 100644 --- a/packages/api/src/services/highlights.ts +++ b/packages/api/src/services/highlights.ts @@ -5,16 +5,20 @@ import { EntityLabel } from '../entity/entity_label' import { Highlight } from '../entity/highlight' import { Label } from '../entity/label' import { homePageURL } from '../env' -import { createPubSubClient, EntityType } from '../pubsub' +import { createPubSubClient, EntityEvent, EntityType } from '../pubsub' import { authTrx } from '../repository' import { highlightRepository } from '../repository/highlight' +import { Merge } from '../util' import { enqueueUpdateHighlight } from '../utils/createTask' import { deepDelete } from '../utils/helpers' import { ItemEvent } from './library_item' const columnToDelete = ['user', 'sharedAt'] as const type ColumnToDeleteType = typeof columnToDelete[number] -export type HighlightEvent = Omit, ColumnToDeleteType> +export type HighlightEvent = Merge< + Omit, ColumnToDeleteType>, + EntityEvent +> export const getHighlightLocation = (patch: string): number | undefined => { const dmp = new diff_match_patch() @@ -51,6 +55,7 @@ export const createHighlight = async ( where: { id: newHighlight.id }, relations: { user: true, + libraryItem: true, }, }) }, @@ -61,7 +66,11 @@ export const createHighlight = async ( const cleanData = deepDelete(newHighlight, columnToDelete) await pubsub.entityCreated( EntityType.HIGHLIGHT, - { id: libraryItemId, highlights: [cleanData] }, + { + id: libraryItemId, + slug: newHighlight.libraryItem.slug, + highlights: [cleanData], + }, userId ) @@ -102,13 +111,18 @@ export const mergeHighlights = async ( where: { id: newHighlight.id }, relations: { user: true, + libraryItem: true, }, }) }) await pubsub.entityCreated( EntityType.HIGHLIGHT, - { id: libraryItemId, highlights: [newHighlight] }, + { + id: libraryItemId, + slug: newHighlight.libraryItem.slug, + highlights: [newHighlight], + }, userId ) @@ -142,7 +156,11 @@ export const updateHighlight = async ( const libraryItemId = updatedHighlight.libraryItem.id await pubsub.entityUpdated( EntityType.HIGHLIGHT, - { id: libraryItemId, highlights: [highlight] } as ItemEvent, + { + id: libraryItemId, + slug: updatedHighlight.libraryItem.slug, + highlights: [highlight], + } as ItemEvent, userId ) diff --git a/packages/api/src/services/integrations/integration.ts b/packages/api/src/services/integrations/integration.ts index 44c95e680..613cae2c3 100644 --- a/packages/api/src/services/integrations/integration.ts +++ b/packages/api/src/services/integrations/integration.ts @@ -1,4 +1,5 @@ -import { LibraryItem, LibraryItemState } from '../../entity/library_item' +import { LibraryItemState } from '../../entity/library_item' +import { ItemEvent } from '../library_item' export interface RetrievedData { url: string @@ -26,5 +27,5 @@ export interface IntegrationClient { auth(state: string): Promise - export(items: LibraryItem[]): Promise + export(items: ItemEvent[]): Promise } diff --git a/packages/api/src/services/integrations/notion.ts b/packages/api/src/services/integrations/notion.ts index ef9b63319..c274f062a 100644 --- a/packages/api/src/services/integrations/notion.ts +++ b/packages/api/src/services/integrations/notion.ts @@ -1,11 +1,11 @@ import { Client } from '@notionhq/client' import axios from 'axios' import { Integration } from '../../entity/integration' -import { LibraryItem } from '../../entity/library_item' import { env } from '../../env' import { Merge } from '../../util' import { logger } from '../../utils/logger' import { getHighlightUrl } from '../highlights' +import { ItemEvent } from '../library_item' import { IntegrationClient } from './integration' type AnnotationColor = @@ -44,7 +44,7 @@ interface NotionPage { } } properties: { - Title: { + Title?: { title: [ { text: { @@ -53,25 +53,28 @@ interface NotionPage { } ] } - Author: { + Author?: { rich_text: Array<{ text: { content: string } }> } - 'Original URL': { + 'Original URL'?: { url: string } - 'Omnivore URL': { + 'Omnivore ID': { + unique_id: string + } + 'Omnivore URL'?: { url: string } - 'Saved At': { + 'Saved At'?: { date: { start: string } } - 'Last Updated': { + 'Last Updated'?: { date: { start: string } @@ -177,7 +180,7 @@ export class NotionClient implements IntegrationClient { } private itemToNotionPage = ( - item: LibraryItem, + item: ItemEvent, settings: Settings, lastSync?: Date | null ): NotionPage => { @@ -200,44 +203,59 @@ export class NotionClient implements IntegrationClient { } : undefined, properties: { - Title: { - title: [ - { - text: { - content: item.title, + Title: item.title + ? { + title: [ + { + text: { + content: item.title, + }, + }, + ], + } + : undefined, + Author: item.author + ? { + rich_text: [ + { + text: { + content: item.author, + }, + }, + ], + } + : undefined, + 'Original URL': item.originalUrl + ? { + url: item.originalUrl, + } + : undefined, + 'Omnivore ID': { + unique_id: item.id, + }, + 'Omnivore URL': item.slug + ? { + url: `${env.client.url}/me/${item.slug}`, + } + : undefined, + 'Saved At': item.savedAt + ? { + date: { + start: (item.savedAt as Date).toISOString(), }, - }, - ], - }, - Author: { - rich_text: [ - { - text: { - content: item.author || 'unknown', + } + : undefined, + 'Last Updated': item.updatedAt + ? { + date: { + start: (item.updatedAt as Date).toISOString(), }, - }, - ], - }, - 'Original URL': { - url: item.originalUrl, - }, - 'Omnivore URL': { - url: `${env.client.url}/me/${item.slug}`, - }, - 'Saved At': { - date: { - start: item.createdAt.toISOString(), - }, - }, - 'Last Updated': { - date: { - start: item.updatedAt.toISOString(), - }, - }, + } + : undefined, Tags: item.labels ? { multi_select: item.labels.map((label) => ({ - name: label.name, + name: label.name || '', })), } : undefined, @@ -246,7 +264,9 @@ export class NotionClient implements IntegrationClient { settings.properties?.includes('highlights') && item.highlights ? item.highlights .filter( - (highlight) => !lastSync || highlight.updatedAt > lastSync // only new highlights + (highlight) => + !lastSync || + (highlight.updatedAt && highlight.updatedAt > lastSync) // only new highlights ) .map((highlight) => ({ paragraph: { @@ -255,7 +275,7 @@ export class NotionClient implements IntegrationClient { text: { content: highlight.quote || '', link: { - url: getHighlightUrl(item.slug, highlight.id), + url: getHighlightUrl(item.slug!, highlight.id), }, }, annotations: { @@ -289,14 +309,14 @@ export class NotionClient implements IntegrationClient { await this.client.pages.create(page) } - private findPage = async (url: string, databaseId: string) => { + private findPage = async (id: string, databaseId: string) => { const response = await this.client.databases.query({ database_id: databaseId, page_size: 1, filter: { - property: 'Omnivore URL', + property: 'Omnivore ID', url: { - equals: url, + equals: id, }, }, }) @@ -307,7 +327,7 @@ export class NotionClient implements IntegrationClient { return null } - export = async (items: LibraryItem[]): Promise => { + export = async (items: ItemEvent[]): Promise => { const settings = this.integrationData?.settings if (!this.integrationData || !settings) { logger.error('Notion integration data not found') diff --git a/packages/api/src/services/integrations/readwise.ts b/packages/api/src/services/integrations/readwise.ts index 0c204b630..b70ef4260 100644 --- a/packages/api/src/services/integrations/readwise.ts +++ b/packages/api/src/services/integrations/readwise.ts @@ -1,7 +1,7 @@ import axios from 'axios' -import { LibraryItem } from '../../entity/library_item' import { logger } from '../../utils/logger' import { getHighlightUrl } from '../highlights' +import { ItemEvent } from '../library_item' import { IntegrationClient } from './integration' interface ReadwiseHighlight { @@ -66,7 +66,7 @@ export class ReadwiseClient implements IntegrationClient { } } - export = async (items: LibraryItem[]): Promise => { + export = async (items: ItemEvent[]): Promise => { let result = true const highlights = items.flatMap(this._itemToReadwiseHighlight) @@ -83,9 +83,7 @@ export class ReadwiseClient implements IntegrationClient { throw new Error('Method not implemented.') } - private _itemToReadwiseHighlight = ( - item: LibraryItem - ): ReadwiseHighlight[] => { + private _itemToReadwiseHighlight = (item: ItemEvent): ReadwiseHighlight[] => { const category = item.siteName === 'Twitter' ? 'tweets' : 'articles' return item.highlights ?.map((highlight) => { @@ -98,8 +96,10 @@ export class ReadwiseClient implements IntegrationClient { text: highlight.quote, title: item.title, author: item.author || undefined, - highlight_url: getHighlightUrl(item.slug, highlight.id), - highlighted_at: new Date(highlight.createdAt).toISOString(), + highlight_url: item.slug + ? getHighlightUrl(item.slug, highlight.id) + : undefined, + highlighted_at: (highlight.createdAt as Date).toISOString(), category, image_url: item.thumbnail || undefined, location_type: 'order', diff --git a/packages/api/src/services/labels.ts b/packages/api/src/services/labels.ts index 84ebc7c23..c2f5c110e 100644 --- a/packages/api/src/services/labels.ts +++ b/packages/api/src/services/labels.ts @@ -2,9 +2,15 @@ import { DeepPartial, FindOptionsWhere, In } from 'typeorm' import { QueryDeepPartialEntity } from 'typeorm/query-builder/QueryPartialEntity' import { EntityLabel, LabelSource } from '../entity/entity_label' import { Label } from '../entity/label' -import { createPubSubClient, EntityType, PubsubClient } from '../pubsub' +import { + createPubSubClient, + EntityEvent, + EntityType, + PubsubClient, +} from '../pubsub' import { authTrx } from '../repository' import { CreateLabelInput, labelRepository } from '../repository/label' +import { Merge } from '../util' import { bulkEnqueueUpdateLabels } from '../utils/createTask' import { deepDelete } from '../utils/helpers' import { logger } from '../utils/logger' @@ -13,7 +19,10 @@ import { findLibraryItemIdsByLabelId, ItemEvent } from './library_item' const columnToDelete = ['description', 'createdAt'] as const type ColumnToDeleteType = typeof columnToDelete[number] -export type LabelEvent = Omit, ColumnToDeleteType> +export type LabelEvent = Merge< + Omit, ColumnToDeleteType>, + EntityEvent +> // const batchGetLabelsFromLinkIds = async ( // linkIds: readonly string[]