Add pocket integration support

packages/api/src/resolvers/integrations/index.ts

@@ -14,7 +14,7 @@ import {
 } from '../../generated/graphql'
 import { getRepository } from '../../entity/utils'
 import { User } from '../../entity/user'
-import { Integration } from '../../entity/integration'
+import { Integration, IntegrationType } from '../../entity/integration'
 import { analytics } from '../../utils/analytics'
 import { env } from '../../env'
 import { getIntegrationService } from '../../services/integrations'
@@ -83,8 +83,11 @@ export const setIntegrationResolver = authorized<
     // save integration
     const integration = await getRepository(Integration).save(integrationToSave)
 
-    if (!integrationToSave.id || integrationToSave.enabled) {
-      // create a task to sync all the pages if new integration or enable integration
+    if (
+      integration.type === IntegrationType.Export &&
+      (!integrationToSave.id || integrationToSave.enabled)
+    ) {
+      // create a task to sync all the pages if new integration or enable integration (export type)
       const taskName = await enqueueSyncWithIntegration(
         user.id,
         input.type as string

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

@@ -101,7 +101,7 @@ export function integrationsServiceRouter() {
           pageId: page.id,
         })
 
-        const synced = await integrationService.exportPages(integration, [page])
+        const synced = await integrationService.export(integration, [page])
         if (!synced) {
           logger.info('failed to sync page', {
             integrationId: integration.id,
@@ -132,10 +132,7 @@ export function integrationsServiceRouter() {
 
           logger.info('syncing pages', { pageIds })
 
-          const synced = await integrationService.exportPages(
-            integration,
-            pages
-          )
+          const synced = await integrationService.export(integration, pages)
           if (!synced) {
             logger.info('failed to sync pages', {
               pageIds,

packages/api/src/services/integrations/integration.ts

@@ -7,16 +7,13 @@ export abstract class IntegrationService {
   validateToken = async (token: string): Promise<boolean> => {
     return Promise.resolve(true)
   }
-  exportPages = async (
+  export = async (
     integration: Integration,
     pages: Page[]
   ): Promise<boolean> => {
     return Promise.resolve(true)
   }
-  importPages = async (
-    integration: Integration,
-    pages: Page[]
-  ): Promise<boolean> => {
-    return Promise.resolve(true)
+  import = async (integration: Integration): Promise<void> => {
+    return Promise.resolve()
   }
 }

packages/api/src/services/integrations/pocket.ts

@@ -1,5 +1,66 @@
 import { IntegrationService } from './integration'
+import { Integration } from '../../entity/integration'
+import axios from 'axios'
+import { env } from '../../env'
+import { PubSub } from '@google-cloud/pubsub'
+
+interface PocketResponse {
+  list: {
+    [key: string]: PocketItem
+  }
+}
+
+interface PocketItem {
+  given_url: string
+}
 
 export class PocketIntegration extends IntegrationService {
   name = 'POCKET'
+  POCKET_API_URL = 'https://getpocket.com/v3'
+  IMPORT_TOPIC = 'importURL'
+
+  retrievePocketData = async (
+    accessToken: string,
+    since: number
+  ): Promise<PocketResponse> => {
+    const url = `${this.POCKET_API_URL}/get`
+    try {
+      const response = await axios.post<PocketResponse>(url, {
+        consumer_key: env.pocket.consumerKey,
+        access_token: accessToken,
+        state: 'all',
+        detailType: 'simple',
+        since,
+      })
+      return response.data
+    } catch (error) {
+      console.log('error retrieving pocket data', error)
+      throw error
+    }
+  }
+
+  import = async (integration: Integration): Promise<void> => {
+    const syncAt = integration.syncedAt
+      ? integration.syncedAt.getTime() / 1000
+      : 0
+    const pocketData = await this.retrievePocketData(integration.token, syncAt)
+    const pocketItems = Object.values(pocketData.list)
+    // publish pocket items to queue
+    const client = new PubSub()
+    await Promise.all(
+      pocketItems.map((item) => {
+        return client
+          .topic(this.IMPORT_TOPIC)
+          .publishMessage({
+            data: JSON.stringify({
+              url: item.given_url,
+            }),
+          })
+          .catch((err) => {
+            console.log('error publishing to pubsub', err)
+            return undefined
+          })
+      })
+    )
+  }
 }

packages/api/src/services/integrations/readwise.ts

@@ -52,7 +52,7 @@ export class ReadwiseIntegration extends IntegrationService {
       return false
     }
   }
-  exportPages = async (
+  export = async (
     integration: Integration,
     pages: Page[]
   ): Promise<boolean> => {

packages/api/src/util.ts

@@ -98,6 +98,10 @@ interface BackendEnv {
   gcp: {
     location: string
   }
+
+  pocket: {
+    consumerKey: string
+  }
 }
 
 /***
@@ -154,6 +158,7 @@ const nullableEnvVars = [
   'AZURE_SPEECH_REGION',
   'GCP_LOCATION',
   'RECOMMENDATION_TASK_HANDLER_URL',
+  'POCKET_CONSUMER_KEY',
 ] // 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 */
@@ -284,6 +289,10 @@ export function getEnv(): BackendEnv {
     location: parse('GCP_LOCATION'),
   }
 
+  const pocket = {
+    consumerKey: parse('POCKET_CONSUMER_KEY'),
+  }
+
   return {
     pg,
     client,
@@ -304,6 +313,7 @@ export function getEnv(): BackendEnv {
     readwise,
     azure,
     gcp,
+    pocket,
   }
 }