Merge pull request #2359 from omnivore-app/fix/pocket-import

fix/pocket import

packages/api/package.json

@@ -49,6 +49,7 @@
     "cookie": "^0.5.0",
     "cookie-parser": "^1.4.5",
     "cors": "^2.8.5",
+    "csv-stringify": "^6.4.0",
     "dataloader": "^2.0.0",
     "diff-match-patch": "^1.0.5",
     "dompurify": "^2.0.17",
@@ -107,6 +108,7 @@
     "@types/chai-string": "^1.4.2",
     "@types/cookie": "^0.4.0",
     "@types/cookie-parser": "^1.4.2",
+    "@types/csv-stringify": "^3.1.0",
     "@types/diff-match-patch": "^1.0.32",
     "@types/dompurify": "^2.0.4",
     "@types/express": "^4.17.7",

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

@@ -1,6 +1,7 @@
 /* eslint-disable @typescript-eslint/no-misused-promises */
 /* eslint-disable @typescript-eslint/no-unsafe-assignment */
 /* eslint-disable @typescript-eslint/no-unsafe-member-access */
+import { stringify } from 'csv-stringify'
 import express from 'express'
 import { DateTime } from 'luxon'
 import { v4 as uuidv4 } from 'uuid'
@@ -220,6 +221,12 @@ export function integrationsServiceRouter() {
       writeStream = file.createWriteStream({
         contentType: 'text/csv',
       })
+      // stringify the data and pipe it to the write_stream
+      const stringifier = stringify({
+        header: false,
+        columns: ['url', 'state', 'labels'],
+      })
+      stringifier.pipe(writeStream)
 
       let hasMore = true
       let offset = 0
@@ -236,11 +243,7 @@ export function integrationsServiceRouter() {
           break
         }
         // write the list of urls, state and labels to the stream
-        const csvData = retrievedData.map((page) => {
-          const { url, state, labels } = page
-          return [url, state, `"[${labels?.join(',') || ''}]"`].join(',')
-        })
-        writeStream.write(csvData.join('\n'))
+        retrievedData.forEach((row) => stringifier.write(row))
 
         hasMore = !!retrieved.hasMore
         offset += retrievedData.length

packages/api/test/routers/integrations.test.ts

@@ -369,8 +369,16 @@ describe('Integrations routers', () => {
           complete: 1,
           list: {
             '123': {
-              given_url: 'https://omnivore.app/pocket-import-test',
+              given_url: 'https://omnivore.app/pocket-import-test,test',
               state: '0',
+              tags: {
+                '1234': {
+                  tag: 'test',
+                },
+                '1235': {
+                  tag: 'new',
+                },
+              },
             },
           },
           since: Date.now() / 1000,
@@ -387,6 +395,7 @@ describe('Integrations routers', () => {
 
     after(async () => {
       sinon.restore()
+      nock.cleanAll()
       await deleteTestIntegrations(user.id, [integration.id])
     })
 

packages/import-handler/package.json

@@ -42,7 +42,6 @@
     "@sentry/serverless": "^7.30.0",
     "@types/express": "^4.17.13",
     "axios": "^1.2.2",
-    "csv-parser": "^3.0.0",
     "dompurify": "^2.4.3",
     "fs-extra": "^11.1.0",
     "glob": "^8.1.0",

packages/import-handler/src/csv.ts

@@ -8,6 +8,22 @@ import { Stream } from 'stream'
 import { ImportContext } from '.'
 import { createMetrics, ImportStatus, updateMetrics } from './metrics'
 
+const parseLabels = (labels: string): string[] => {
+  try {
+    // labels follows format: "[""label1"",""label2""]"
+    return JSON.parse(labels) as string[]
+  } catch (error) {
+    console.debug('invalid labels format', labels)
+
+    // labels follows format: "[label1,label2]"
+    return labels
+      .slice(1, -1)
+      .split(',')
+      .map((l) => l.trim())
+      .filter((l) => l !== '')
+  }
+}
+
 export const importCsv = async (ctx: ImportContext, stream: Stream) => {
   // create metrics in redis
   await createMetrics(ctx.redisClient, ctx.userId, ctx.taskId, 'csv-importer')
@@ -18,15 +34,7 @@ export const importCsv = async (ctx: ImportContext, stream: Stream) => {
     try {
       const url = new URL(row[0])
       const state = row.length > 1 && row[1] ? row[1] : undefined
-      // labels follows format: "[label1,label2]"
-      const labels =
-        row.length > 2
-          ? (row[2] as string)
-              .slice(1, -1)
-              .split(',')
-              .map((l) => l.trim())
-              .filter((l) => l !== '')
-          : undefined
+      const labels = row.length > 2 ? parseLabels(row[2]) : undefined
 
       // update total counter
       await updateMetrics(

packages/import-handler/src/index.ts

@@ -127,7 +127,12 @@ const createEmailCloudTask = async (userId: string, payload: unknown) => {
     Cookie: `auth=${authToken}`,
   }
 
-  return createCloudTask(emailUserUrl(), payload, headers)
+  return createCloudTask(
+    emailUserUrl(),
+    payload,
+    headers,
+    'omnivore-email-queue'
+  )
 }
 
 const sendImportFailedEmail = async (userId: string) => {

packages/import-handler/src/task.ts

@@ -16,9 +16,9 @@ export const CONTENT_FETCH_URL = process.env.CONTENT_FETCH_GCF_URL
 export const createCloudTask = async (
   taskHandlerUrl: string | undefined,
   payload: unknown,
-  requestHeaders?: Record<string, string>
+  requestHeaders?: Record<string, string>,
+  queue = 'omnivore-import-queue'
 ) => {
-  const queue = 'omnivore-import-queue'
   const location = process.env.GCP_LOCATION
   const project = process.env.GCP_PROJECT_ID
 

packages/import-handler/test/csv/data/complex.csv

@@ -1,3 +1,3 @@
-"https://omnivore.app",ARCHIVED,"[test]"
-"https://google.com",SUCCEEDED,"[test,development]"
+"https://omnivore.app",ARCHIVED,"[""test""]"
+"https://google.com",SUCCEEDED,"[""test"",""development""]"
 https://test.com,SUCCEEDED,"[test, development]"

packages/puppeteer-parse/index.js

@@ -461,6 +461,7 @@ async function fetchContent(req, res) {
     // mark import failed on the last failed retry
     const retryCount = req.headers['x-cloudtasks-taskretrycount'];
     if (retryCount == MAX_RETRY_COUNT) {
+      console.debug('max retry count reached');
       importStatus = importStatus || 'failed';
     }
 

yarn.lock

@@ -8111,6 +8111,13 @@
   resolved "https://registry.yarnpkg.com/@types/cors/-/cors-2.8.12.tgz#6b2c510a7ad7039e98e7b8d3d6598f4359e5c080"
   integrity sha512-vt+kDhq/M2ayberEtJcIN/hxXy1Pk+59g2FV/ZQceeaTyCtCucjL2Q7FXlFjtWn4n15KCr1NE2lNNFhp0lEThw==
 
+"@types/csv-stringify@^3.1.0":
+  version "3.1.0"
+  resolved "https://registry.yarnpkg.com/@types/csv-stringify/-/csv-stringify-3.1.0.tgz#4c172ef462740e584a5bfe66ea78b67759f7bb32"
+  integrity sha512-jNRWx49wIc9UjJXukCaQt8iZRjyzDiEC1CGAAIZsydECWl5xM9oq4pSc5+Jhl4oATrRr+eGA9Vf0y9duDbKAvg==
+  dependencies:
+    csv-stringify "*"
+
 "@types/debug@^4.0.0", "@types/debug@^4.1.0":
   version "4.1.7"
   resolved "https://registry.yarnpkg.com/@types/debug/-/debug-4.1.7.tgz#7cc0ea761509124709b8b2d1090d8f6c17aadb82"
@@ -12682,12 +12689,10 @@ csstype@^3.0.2, csstype@^3.0.4:
   resolved "https://registry.yarnpkg.com/csstype/-/csstype-3.0.8.tgz#d2266a792729fb227cd216fb572f43728e1ad340"
   integrity sha512-jXKhWqXPmlUeoQnF/EhTtTl4C9SnrxSH/jZUih3jmO6lBKr99rP3/+FmrMj4EFpOXzMtXHAZkd3x0E6h6Fgflw==
 
-csv-parser@^3.0.0:
-  version "3.0.0"
-  resolved "https://registry.yarnpkg.com/csv-parser/-/csv-parser-3.0.0.tgz#b88a6256d79e090a97a1b56451f9327b01d710e7"
-  integrity sha512-s6OYSXAK3IdKqYO33y09jhypG/bSDHPuyCme/IdEHfWpLf/jKcpitVFyOC6UemgGk8v7Q5u2XE0vvwmanxhGlQ==
-  dependencies:
-    minimist "^1.2.0"
+csv-stringify@*, csv-stringify@^6.4.0:
+  version "6.4.0"
+  resolved "https://registry.yarnpkg.com/csv-stringify/-/csv-stringify-6.4.0.tgz#6d006dca9194700e44f9fbc541bee8bbbd4f459c"
+  integrity sha512-HQsw0QXiN5fdlO+R8/JzCZnR3Fqp8E87YVnhHlaPtNGJjt6ffbV0LpOkieIb1x6V1+xt878IYq77SpXHWAqKkA==
 
 cyclist@^1.0.1:
   version "1.0.1"