create an API to create a cloud task for each active integration when the cronjob triggers

packages/api/src/routers/auth/jwt_helpers.ts

@@ -85,3 +85,14 @@ export function suggestedUsername(name: string): string {
   const suffix = Math.floor(Math.random() * 10000)
   return `${prefix}${suffix}`
 }
+
+export async function createWebAuthTokenWithPayload(
+  payload: Record<string, unknown>
+): Promise<string | undefined> {
+  try {
+    const authToken = await signToken(payload, env.server.jwtSecret)
+    return authToken as string
+  } catch {
+    return undefined
+  }
+}

packages/api/src/routers/svc/integrations.ts

@@ -5,9 +5,10 @@ import { stringify } from 'csv-stringify'
 import express from 'express'
 import { DateTime } from 'luxon'
 import { v4 as uuidv4 } from 'uuid'
-import { IntegrationType } from '../../entity/integration'
+import { Integration, IntegrationType } from '../../entity/integration'
 import { LibraryItem } from '../../entity/library_item'
 import { EntityType, readPushSubscription } from '../../pubsub'
+import { getRepository } from '../../repository'
 import { Claims } from '../../resolvers/types'
 import {
   findIntegration,
@@ -19,9 +20,11 @@ import {
   searchLibraryItems,
 } from '../../services/library_item'
 import { getClaimsByToken } from '../../utils/auth'
+import { enqueueExportToIntegration } from '../../utils/createTask'
 import { logger } from '../../utils/logger'
 import { DateFilter } from '../../utils/search'
 import { createGCSFile } from '../../utils/uploads'
+import { createWebAuthTokenWithPayload } from '../auth/jwt_helpers'
 
 export interface Message {
   type?: EntityType
@@ -41,6 +44,63 @@ const isImportEvent = (event: any): event is ImportEvent =>
 export function integrationsServiceRouter() {
   const router = express.Router()
 
+  router.post('/export', async (req, res) => {
+    logger.info('start to sync with integration')
+
+    try {
+      const { message: msgStr, expired } = readPushSubscription(req)
+      if (!msgStr) {
+        return res.status(200).send('Bad Request')
+      }
+
+      if (expired) {
+        logger.info('discarding expired message')
+        return res.status(200).send('Expired')
+      }
+
+      // find all active integrations
+      const integrations = await getRepository(Integration).find({
+        where: {
+          enabled: true,
+          type: IntegrationType.Export,
+        },
+        relations: ['user'],
+      })
+
+      // create a task to sync with each integration
+      await Promise.all(
+        integrations.map(async (integration) => {
+          const exp = Math.floor(Date.now() / 1000) + 60 * 60 * 24 // 1 day
+          const authToken = await createWebAuthTokenWithPayload({
+            uid: integration.user.id,
+            exp,
+            token: integration.token,
+          })
+
+          if (!authToken) {
+            logger.error('failed to create auth token', {
+              integrationId: integration.id,
+            })
+            return
+          }
+
+          const syncAt = integration.syncedAt?.getTime() || 0
+          return enqueueExportToIntegration(
+            integration.id,
+            integration.name,
+            syncAt,
+            authToken
+          )
+        })
+      )
+    } catch (err) {
+      logger.error('sync with integrations failed', err)
+      return res.status(500).send(err)
+    }
+
+    res.status(200).send('OK')
+  })
+
   router.post('/:integrationName/:action', async (req, res) => {
     logger.info('start to sync with integration', {
       action: req.params.action,

packages/api/src/util.ts

@@ -70,6 +70,8 @@ interface BackendEnv {
     recommendationTaskHandlerUrl: string
     thumbnailTaskHandlerUrl: string
     rssFeedTaskHandlerUrl: string
+    integrationExporterUrl: string
+    integrationImporterUrl: string
   }
   fileUpload: {
     gcsUploadBucket: string
@@ -163,6 +165,8 @@ const nullableEnvVars = [
   'SENDGRID_VERIFICATION_TEMPLATE_ID',
   'REMINDER_TASK_HANDLER_URL',
   'TRUST_PROXY',
+  'INTEGRATION_EXPORTER_URL',
+  'INTEGRATION_IMPORTER_URL',
 ] // Allow some vars to be null/empty
 
 /* If not in GAE and Prod/QA/Demo env (f.e. on localhost/dev env), allow following env vars to be null */
@@ -253,6 +257,8 @@ export function getEnv(): BackendEnv {
     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'),
   }
   const imageProxy = {
     url: parse('IMAGE_PROXY_URL'),

packages/api/src/utils/createTask.ts

@@ -543,6 +543,57 @@ export const enqueueImportFromIntegration = async (
   return createdTasks[0].name
 }
 
+export const enqueueExportToIntegration = async (
+  integrationId: string,
+  integrationName: string,
+  syncAt: number, // unix timestamp in milliseconds
+  authToken: string
+): Promise<string> => {
+  const { GOOGLE_CLOUD_PROJECT } = process.env
+  const payload = {
+    integrationId,
+    integrationName,
+    syncAt,
+  }
+
+  const headers = {
+    Cookie: `auth=${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.integrationTaskHandlerUrl) {
+      // 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,
+  })
+
+  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 enqueueThumbnailTask = async (
   userId: string,
   slug: string