From df7e1fd269a1473aae8561caf8e0c3d3ede5e5e5 Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Wed, 24 May 2023 16:57:42 +0800
Subject: [PATCH 01/12] fix: catch scrapingbee api error

---
 packages/puppeteer-parse/index.js | 33 ++++++++++++++++++-------------
 1 file changed, 19 insertions(+), 14 deletions(-)

diff --git a/packages/puppeteer-parse/index.js b/packages/puppeteer-parse/index.js
index 960a8e735..49797c39a 100644
--- a/packages/puppeteer-parse/index.js
+++ b/packages/puppeteer-parse/index.js
@@ -59,18 +59,23 @@ const userAgentForUrl = (url) => {
 };
 
 const fetchContentWithScrapingBee = async (url) => {
-  const response = await axios.get('https://app.scrapingbee.com/api/v1', {
-    params: {
-      'api_key':  process.env.SCRAPINGBEE_API_KEY,
-      'url': url,
-      'render_js': 'false',
-      'premium_proxy': 'true',
-      'country_code':'us'
-    }
-  })
-
-  const dom = parseHTML(response.data).document;
-  return { title: dom.title, domContent: dom.documentElement.outerHTML, url: url }
+  try {
+    const response = await axios.get('https://app.scrapingbee.com/api/v1', {
+      params: {
+        'api_key':  process.env.SCRAPINGBEE_API_KEY,
+        'url': url,
+        'render_js': 'false',
+        'premium_proxy': 'true',
+        'country_code':'us'
+      }
+    })
+  
+    const dom = parseHTML(response.data).document;
+    return { title: dom.title, domContent: dom.documentElement.outerHTML, url }
+  } catch (e) {
+    console.log('error fetching with scrapingbee', e)
+    return { title: '', domContent: '', url }
+  }
 }
 
 const enableJavascriptForUrl = (url) => {
@@ -350,8 +355,8 @@ async function fetchContent(req, res) {
 
     // fallback to scrapingbee
     const sbResult = await fetchContentWithScrapingBee(url);
-    const sbUrl = finalUrl || sbResult.url;
     const content = sbResult.domContent;
+    const title = sbResult.title;
     logRecord.fetchContentTime = Date.now() - functionStartTime;
 
     let readabilityResult = null;
@@ -359,7 +364,7 @@ async function fetchContent(req, res) {
       let document = parseHTML(content).document;
 
       // preParse content
-      const preParsedDom = await preParseContent(sbUrl, document)
+      const preParsedDom = await preParseContent(url, document)
       if (preParsedDom) {
         document = preParsedDom
       }

From 8ddde61285963a83887d69410c59c1c9800e08fe Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Wed, 24 May 2023 18:00:49 +0800
Subject: [PATCH 02/12] fix: increase twitter-handler timeout to 60 seconds

---
 packages/content-handler/src/websites/twitter-handler.ts | 7 ++++---
 packages/puppeteer-parse/index.js                        | 4 ++--
 2 files changed, 6 insertions(+), 5 deletions(-)

diff --git a/packages/content-handler/src/websites/twitter-handler.ts b/packages/content-handler/src/websites/twitter-handler.ts
index 16fa054c2..832cc7e13 100644
--- a/packages/content-handler/src/websites/twitter-handler.ts
+++ b/packages/content-handler/src/websites/twitter-handler.ts
@@ -1,9 +1,9 @@
-import { ContentHandler, PreHandleResult } from '../content-handler'
 import axios from 'axios'
-import { DateTime } from 'luxon'
-import _ from 'underscore'
 import { truncate } from 'lodash'
+import { DateTime } from 'luxon'
 import { Browser, BrowserContext } from 'puppeteer-core'
+import _ from 'underscore'
+import { ContentHandler, PreHandleResult } from '../content-handler'
 
 interface TweetIncludes {
   users: {
@@ -219,6 +219,7 @@ const getTweetIds = async (
 
     await page.goto(pageURL, {
       waitUntil: 'networkidle0',
+      timeout: 60000, // 60 seconds
     })
 
     return (await page.evaluate(async (author) => {
diff --git a/packages/puppeteer-parse/index.js b/packages/puppeteer-parse/index.js
index 49797c39a..f4e5e55c9 100644
--- a/packages/puppeteer-parse/index.js
+++ b/packages/puppeteer-parse/index.js
@@ -389,9 +389,9 @@ async function fetchContent(req, res) {
       await context.close();
     }
     console.info(`parse-page`, logRecord);
-  }
 
-  return res.sendStatus(200);
+    res.sendStatus(200);
+  }
 }
 
 function validateUrlString(url) {

From d69e732579fb3c6a1ab194df72fdc1ff8ca72368 Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Wed, 24 May 2023 20:29:35 +0800
Subject: [PATCH 03/12] fix: dedupe url if failed to save

---
 packages/api/src/utils/parser.ts              | 21 ++++++++++++++-----
 .../src/websites/twitter-handler.ts           |  3 ++-
 packages/puppeteer-parse/index.js             | 18 +++++++---------
 3 files changed, 26 insertions(+), 16 deletions(-)

diff --git a/packages/api/src/utils/parser.ts b/packages/api/src/utils/parser.ts
index 6d255677a..e48ca7326 100644
--- a/packages/api/src/utils/parser.ts
+++ b/packages/api/src/utils/parser.ts
@@ -207,6 +207,16 @@ export const parsePreparedContent = async (
   let highlightData = undefined
   const { document, pageInfo } = preparedDocument
 
+  if (!document) {
+    console.log('No document')
+    return {
+      canonicalUrl: url,
+      parsedContent: null,
+      domContent: '',
+      pageType: PageType.Unknown,
+    }
+  }
+
   // Checking for content type acceptance or if there are no contentType
   // at all (backward extension versions compatibility)
   if (
@@ -222,14 +232,15 @@ export const parsePreparedContent = async (
     }
   }
 
-  let dom = parseHTML(document).document
+  let dom: Document | null = null
 
   try {
+    dom = parseHTML(document).document
+
     if (!article) {
       // Attempt to parse the article
       // preParse content
-      const preParsedDom = await preParseContent(url, dom)
-      preParsedDom && (dom = preParsedDom)
+      dom = (await preParseContent(url, dom)) || dom
 
       article = await getReadabilityResult(url, document, dom, isNewsletter)
     }
@@ -260,7 +271,7 @@ export const parsePreparedContent = async (
         codeBlocks.forEach((e) => {
           if (e.textContent) {
             const att = hljs.highlightAuto(e.textContent)
-            const code = dom.createElement('code')
+            const code = articleDom.createElement('code')
             const langClass =
               `hljs language-${att.language}` +
               (att.second_best?.language
@@ -356,7 +367,7 @@ export const parsePreparedContent = async (
     domContent: document,
     parsedContent: article,
     canonicalUrl,
-    pageType: parseOriginalContent(dom),
+    pageType: dom ? parseOriginalContent(dom) : PageType.Unknown,
     highlightData,
   }
 }
diff --git a/packages/content-handler/src/websites/twitter-handler.ts b/packages/content-handler/src/websites/twitter-handler.ts
index 832cc7e13..c1878e498 100644
--- a/packages/content-handler/src/websites/twitter-handler.ts
+++ b/packages/content-handler/src/websites/twitter-handler.ts
@@ -288,7 +288,8 @@ const getTweetIds = async (
       return Array.from(ids)
     }, author)) as string[]
   } catch (error) {
-    console.log(error)
+    console.error('Error getting tweets', error)
+
     return []
   } finally {
     if (context) {
diff --git a/packages/puppeteer-parse/index.js b/packages/puppeteer-parse/index.js
index f4e5e55c9..352137ac8 100644
--- a/packages/puppeteer-parse/index.js
+++ b/packages/puppeteer-parse/index.js
@@ -73,8 +73,9 @@ const fetchContentWithScrapingBee = async (url) => {
     const dom = parseHTML(response.data).document;
     return { title: dom.title, domContent: dom.documentElement.outerHTML, url }
   } catch (e) {
-    console.log('error fetching with scrapingbee', e)
-    return { title: '', domContent: '', url }
+    console.error('error fetching with scrapingbee', e.message)
+
+    return { title: url, domContent: '', url }
   }
 }
 
@@ -325,19 +326,16 @@ async function fetchContent(req, res) {
 
       let readabilityResult = null;
       if (content) {
-        let document = parseHTML(content).document;
+        const document = parseHTML(content).document;
 
         // preParse content
-        const preParsedDom = await preParseContent(url, document)
-        if (preParsedDom) {
-          document = preParsedDom
-        }
+        const preParsedDom = (await preParseContent(url, document)) || document;
 
-        readabilityResult = await getReadabilityResult(url, document);
+        readabilityResult = await getReadabilityResult(url, preParsedDom);
       }
 
       const apiResponse = await sendSavePageMutation(userId, {
-        url: finalUrl,
+        url,
         clientRequestId: articleSavingRequestId,
         title,
         originalContent: content,
@@ -373,7 +371,7 @@ async function fetchContent(req, res) {
     }
 
     const apiResponse = await sendSavePageMutation(userId, {
-      url: finalUrl,
+      url,
       clientRequestId: articleSavingRequestId,
       title,
       originalContent: content,

From 19facec4e281d46b16d18a0389477c8f38d5e73b Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Thu, 25 May 2023 12:15:05 +0800
Subject: [PATCH 04/12] feat: add importer metrics collector

---
 packages/import-handler/package.json          |  7 +-
 .../src/luaScripts/updateMetrics.lua          | 31 ++++++
 packages/import-handler/src/metrics.ts        | 95 +++++++++++++++++++
 packages/import-handler/src/redis.ts          | 37 ++++++++
 packages/puppeteer-parse/index.js             |  7 +-
 yarn.lock                                     | 51 ++++++++++
 6 files changed, 224 insertions(+), 4 deletions(-)
 create mode 100644 packages/import-handler/src/luaScripts/updateMetrics.lua
 create mode 100644 packages/import-handler/src/metrics.ts
 create mode 100644 packages/import-handler/src/redis.ts

diff --git a/packages/import-handler/package.json b/packages/import-handler/package.json
index c425e3764..d13ce1623 100644
--- a/packages/import-handler/package.json
+++ b/packages/import-handler/package.json
@@ -12,9 +12,10 @@
     "test": "yarn mocha -r ts-node/register --config mocha-config.json",
     "lint": "eslint src --ext ts,js,tsx,jsx",
     "compile": "tsc",
-    "build": "tsc",
+    "build": "tsc && yarn copy-files",
     "start": "functions-framework --target=importHandler",
-    "dev": "concurrently \"tsc -w\" \"nodemon --watch ./build/ --exec npm run start\""
+    "dev": "concurrently \"tsc -w\" \"nodemon --watch ./build/ --exec npm run start\"",
+    "copy-files": "copyfiles -u 1 src/**/*.lua build/src"
   },
   "devDependencies": {
     "@types/chai": "^4.3.4",
@@ -28,6 +29,7 @@
     "@types/unzip-stream": "^0.3.1",
     "@types/urlsafe-base64": "^1.0.28",
     "@types/uuid": "^9.0.0",
+    "copyfiles": "^2.4.1",
     "eslint-plugin-prettier": "^4.0.0"
   },
   "dependencies": {
@@ -46,6 +48,7 @@
     "jsonwebtoken": "^8.5.1",
     "linkedom": "^0.14.21",
     "nodemon": "^2.0.15",
+    "redis": "^4.6.6",
     "unzip-stream": "^0.3.1",
     "urlsafe-base64": "^1.0.0",
     "uuid": "^9.0.0"
diff --git a/packages/import-handler/src/luaScripts/updateMetrics.lua b/packages/import-handler/src/luaScripts/updateMetrics.lua
new file mode 100644
index 000000000..94c45e1ab
--- /dev/null
+++ b/packages/import-handler/src/luaScripts/updateMetrics.lua
@@ -0,0 +1,31 @@
+local key = tostring(KEYS[1]);
+local status = tostring(ARGV[1]);
+local timestamp = tonumber(ARGV[2]);
+
+-- increment the status counter
+redis.call('HINCRBY', key, status, 1);
+
+if (status == "imported" or status == "failed") then
+  -- get the current metrics
+  local bulk = redis.call('HGETALL', key);
+  -- get the total, imported and failed counters
+  local result = {}
+	local nextkey
+	for i, v in ipairs(bulk) do
+		if i % 2 == 1 then
+			nextkey = v
+		else
+			result[nextkey] = v
+		end
+	end
+	
+  local imported = tonumber(result['imported']) or 0;
+  local failed = tonumber(result['failed']) or 0;
+  local total = tonumber(result['total']) or 0;
+  local state = tonumber(result['state']) or 0;
+  if (state == 0 and imported + failed >= total) then
+    -- all the records have been processed
+    -- update the metrics
+    redis.call('HSET', key, 'end_time', timestamp, 'state', 1);
+  end
+end
diff --git a/packages/import-handler/src/metrics.ts b/packages/import-handler/src/metrics.ts
new file mode 100644
index 000000000..7be5edff5
--- /dev/null
+++ b/packages/import-handler/src/metrics.ts
@@ -0,0 +1,95 @@
+import { createClient } from 'redis'
+import { lua } from './redis'
+
+// explicitly create the return type of RedisClient
+type RedisClient = ReturnType<typeof createClient>
+
+enum ImportStatus {
+  STARTED = 'started',
+  INVALID = 'invalid',
+  IMPORTED = 'imported',
+  FAILED = 'failed',
+  TOTAL = 'total',
+}
+
+enum ImportTaskState {
+  STARTED,
+  FINISHED,
+}
+
+interface ImportMetrics {
+  started: number
+  invalid: number
+  imported: number
+  failed: number
+  total: number
+  importer: string
+  state: ImportTaskState
+  startTime: number
+  endTime: number
+}
+
+export const startImport = async (
+  redisClient: RedisClient,
+  userId: string,
+  taskId: string,
+  importer: string
+) => {
+  const key = `import:${userId}:${taskId}`
+  try {
+    // set multiple fields
+    await redisClient.hSet(key, {
+      ['start_time']: Date.now(), // unix timestamp in seconds
+      ['importer']: importer,
+      ['state']: ImportTaskState.STARTED,
+    })
+  } catch (error) {
+    console.error('Redis Error', error)
+  }
+}
+
+export const updateMetrics = async (
+  redisClient: RedisClient,
+  userId: string,
+  taskId: string,
+  status: ImportStatus
+) => {
+  const key = `import:${userId}:${taskId}`
+
+  try {
+    // use lua script to increment hash field
+    await redisClient.evalSha(lua.sha, {
+      keys: [key],
+      arguments: [status, Date.now().toString()],
+    })
+  } catch (error) {
+    console.error('Redis Error', error)
+  }
+}
+
+export const getMetrics = async (
+  redisClient: RedisClient,
+  userId: string,
+  taskId: string
+): Promise<ImportMetrics | null> => {
+  const key = `import:${userId}:${taskId}`
+  try {
+    const metrics = await redisClient.hGetAll(key)
+
+    return {
+      // convert to integer
+      started: parseInt(metrics.started, 10),
+      invalid: parseInt(metrics.invalid, 10),
+      imported: parseInt(metrics.imported, 10),
+      failed: parseInt(metrics.failed, 10),
+      total: parseInt(metrics.total, 10),
+      importer: metrics.importer,
+      state: parseInt(metrics.state, 10),
+      startTime: parseInt(metrics.start_time, 10),
+      endTime: parseInt(metrics.end_time, 10),
+    }
+  } catch (error) {
+    console.error('Redis Error', error)
+    return null
+  }
+}
diff --git a/packages/import-handler/src/redis.ts b/packages/import-handler/src/redis.ts
new file mode 100644
index 000000000..3804c36ed
--- /dev/null
+++ b/packages/import-handler/src/redis.ts
@@ -0,0 +1,37 @@
+import fs from 'fs'
+import { createClient } from 'redis'
+
+// load lua script
+export const lua = {
+  script: fs.readFileSync('./luaScripts/updateMetrics.lua', 'utf8'),
+  sha: '',
+}
+
+export const createRedisClient = async (url?: string, cert?: string) => {
+  const redisClient = createClient({
+    url,
+    socket: {
+      tls: url?.startsWith('rediss://'), // rediss:// is the protocol for TLS
+      cert: cert?.replace(/\\n/g, '\n'), // replace \n with new line
+      rejectUnauthorized: false, // for self-signed certs
+      connectTimeout: 10000, // 10 seconds
+      reconnectStrategy(retries: number): number | Error {
+        if (retries > 10) {
+          return new Error('Retries exhausted')
+        }
+        return 1000
+      },
+    },
+  })
+
+  redisClient.on('error', (err) => console.error('Redis Client Error', err))
+
+  await redisClient.connect()
+  console.log('Redis Client Connected:', url)
+
+  // load script to redis
+  lua.sha = await redisClient.scriptLoad(lua.script)
+  console.log('Redis Lua Script Loaded', lua.sha)
+
+  return redisClient
+}
diff --git a/packages/puppeteer-parse/index.js b/packages/puppeteer-parse/index.js
index 352137ac8..b25498790 100644
--- a/packages/puppeteer-parse/index.js
+++ b/packages/puppeteer-parse/index.js
@@ -224,7 +224,7 @@ const sendSavePageMutation = async (userId, input) => {
           }
     }`,
     variables: {
-      input: Object.assign({}, input , { source: 'puppeteer-parse' }),
+      input,
     },
   });
 
@@ -256,13 +256,14 @@ async function fetchContent(req, res) {
   const articleSavingRequestId = (req.query ? req.query.saveRequestId : undefined) || (req.body ? req.body.saveRequestId : undefined);
   const state = req.body.state
   const labels = req.body.labels
+  const source = req.body.source || 'puppeteer-parse';
 
   let logRecord = {
     url,
     userId,
     articleSavingRequestId,
     labels: {
-      source: 'parseContent',
+      source,
     },
     state,
     labelsToAdd: labels
@@ -342,6 +343,7 @@ async function fetchContent(req, res) {
         parseResult: readabilityResult,
         state,
         labels,
+        source,
       });
 
       logRecord.totalTime = Date.now() - functionStartTime;
@@ -378,6 +380,7 @@ async function fetchContent(req, res) {
       parseResult: readabilityResult,
       state,
       labels,
+      source,
     });
 
     logRecord.totalTime = Date.now() - functionStartTime;
diff --git a/yarn.lock b/yarn.lock
index 339090cfe..5ef541069 100644
--- a/yarn.lock
+++ b/yarn.lock
@@ -5431,6 +5431,11 @@
   resolved "https://registry.yarnpkg.com/@redis/bloom/-/bloom-1.0.2.tgz#42b82ec399a92db05e29fffcdfd9235a5fc15cdf"
   integrity sha512-EBw7Ag1hPgFzdznK2PBblc1kdlj5B5Cw3XwI9/oG7tSn85/HKy3X9xHy/8tm/eNXJYHLXHJL/pkwBpFMVVefkw==
 
+"@redis/bloom@1.2.0":
+  version "1.2.0"
+  resolved "https://registry.yarnpkg.com/@redis/bloom/-/bloom-1.2.0.tgz#d3fd6d3c0af3ef92f26767b56414a370c7b63b71"
+  integrity sha512-HG2DFjYKbpNmVXsa0keLHp/3leGJz1mjh09f2RLGGLQZzSHpkmZWuwJbAvo3QcRY8p80m5+ZdXZdYOSBLlp7Cg==
+
 "@redis/client@1.3.0":
   version "1.3.0"
   resolved "https://registry.yarnpkg.com/@redis/client/-/client-1.3.0.tgz#c62ccd707f16370a2dc2f9e158a28b7da049fa77"
@@ -5440,11 +5445,25 @@
     generic-pool "3.8.2"
     yallist "4.0.0"
 
+"@redis/client@1.5.7":
+  version "1.5.7"
+  resolved "https://registry.yarnpkg.com/@redis/client/-/client-1.5.7.tgz#92cc5c98c76f189e37d24f0e1e17e104c6af17d4"
+  integrity sha512-gaOBOuJPjK5fGtxSseaKgSvjiZXQCdLlGg9WYQst+/GRUjmXaiB5kVkeQMRtPc7Q2t93XZcJfBMSwzs/XS9UZw==
+  dependencies:
+    cluster-key-slot "1.1.2"
+    generic-pool "3.9.0"
+    yallist "4.0.0"
+
 "@redis/graph@1.0.1":
   version "1.0.1"
   resolved "https://registry.yarnpkg.com/@redis/graph/-/graph-1.0.1.tgz#eabc58ba99cd70d0c907169c02b55497e4ec8a99"
   integrity sha512-oDE4myMCJOCVKYMygEMWuriBgqlS5FqdWerikMoJxzmmTUErnTRRgmIDa2VcgytACZMFqpAOWDzops4DOlnkfQ==
 
+"@redis/graph@1.1.0":
+  version "1.1.0"
+  resolved "https://registry.yarnpkg.com/@redis/graph/-/graph-1.1.0.tgz#cc2b82e5141a29ada2cce7d267a6b74baa6dd519"
+  integrity sha512-16yZWngxyXPd+MJxeSr0dqh2AIOi8j9yXKcKCwVaKDbH3HTuETpDVPcLujhFYVPtYrngSco31BUcSa9TH31Gqg==
+
 "@redis/json@1.0.4":
   version "1.0.4"
   resolved "https://registry.yarnpkg.com/@redis/json/-/json-1.0.4.tgz#f372b5f93324e6ffb7f16aadcbcb4e5c3d39bda1"
@@ -5455,11 +5474,21 @@
   resolved "https://registry.yarnpkg.com/@redis/search/-/search-1.1.0.tgz#7abb18d431f27ceafe6bcb4dd83a3fa67e9ab4df"
   integrity sha512-NyFZEVnxIJEybpy+YskjgOJRNsfTYqaPbK/Buv6W2kmFNaRk85JiqjJZA5QkRmWvGbyQYwoO5QfDi2wHskKrQQ==
 
+"@redis/search@1.1.2":
+  version "1.1.2"
+  resolved "https://registry.yarnpkg.com/@redis/search/-/search-1.1.2.tgz#6a8f66ba90812d39c2457420f859ce8fbd8f3838"
+  integrity sha512-/cMfstG/fOh/SsE+4/BQGeuH/JJloeWuH+qJzM8dbxuWvdWibWAOAHHCZTMPhV3xIlH4/cUEIA8OV5QnYpaVoA==
+
 "@redis/time-series@1.0.3":
   version "1.0.3"
   resolved "https://registry.yarnpkg.com/@redis/time-series/-/time-series-1.0.3.tgz#4cfca8e564228c0bddcdf4418cba60c20b224ac4"
   integrity sha512-OFp0q4SGrTH0Mruf6oFsHGea58u8vS/iI5+NpYdicaM+7BgqBZH8FFvNZ8rYYLrUO/QRqMq72NpXmxLVNcdmjA==
 
+"@redis/time-series@1.0.4":
+  version "1.0.4"
+  resolved "https://registry.yarnpkg.com/@redis/time-series/-/time-series-1.0.4.tgz#af85eb080f6934580e4d3b58046026b6c2b18717"
+  integrity sha512-ThUIgo2U/g7cCuZavucQTQzA9g9JbDDY2f64u3AbAoz/8vE2lt2U37LamDUVChhaDA3IRT9R6VvJwqnUfTJzng==
+
 "@remusao/guess-url-type@^1.1.2":
   version "1.2.1"
   resolved "https://registry.yarnpkg.com/@remusao/guess-url-type/-/guess-url-type-1.2.1.tgz#b3e7c32abdf98d0fb4f93cc67cad580b5fe4ba57"
@@ -11812,6 +11841,11 @@ cluster-key-slot@1.1.0:
   resolved "https://registry.yarnpkg.com/cluster-key-slot/-/cluster-key-slot-1.1.0.tgz#30474b2a981fb12172695833052bc0d01336d10d"
   integrity sha512-2Nii8p3RwAPiFwsnZvukotvow2rIHM+yQ6ZcBXGHdniadkYGZYiGmkHJIbZPIV9nfv7m/U1IPMVVcAhoWFeklw==
 
+cluster-key-slot@1.1.2:
+  version "1.1.2"
+  resolved "https://registry.yarnpkg.com/cluster-key-slot/-/cluster-key-slot-1.1.2.tgz#88ddaa46906e303b5de30d3153b7d9fe0a0c19ac"
+  integrity sha512-RMr0FhtfXemyinomL4hrWcYJxmX6deFdCxpJzhDttxgO1+bcCnkk+9drydLVDmAMG7NE6aN/fl4F7ucU/90gAA==
+
 cmd-shim@^4.1.0:
   version "4.1.0"
   resolved "https://registry.yarnpkg.com/cmd-shim/-/cmd-shim-4.1.0.tgz#b3a904a6743e9fede4148c6f3800bf2a08135bdd"
@@ -15320,6 +15354,11 @@ generic-pool@3.8.2:
   resolved "https://registry.yarnpkg.com/generic-pool/-/generic-pool-3.8.2.tgz#aab4f280adb522fdfbdc5e5b64d718d3683f04e9"
   integrity sha512-nGToKy6p3PAbYQ7p1UlWl6vSPwfwU6TMSWK7TTu+WUY4ZjyZQGniGGt2oNVvyNSpyZYSB43zMXVLcBm08MTMkg==
 
+generic-pool@3.9.0:
+  version "3.9.0"
+  resolved "https://registry.yarnpkg.com/generic-pool/-/generic-pool-3.9.0.tgz#36f4a678e963f4fdb8707eab050823abc4e8f5e4"
+  integrity sha512-hymDOu5B53XvN4QT9dBmZxPX4CWhBPPLguTZ9MMFeFa/Kg0xWVfylOVNlJji/E7yTZWFd/q9GO5TxDLq156D7g==
+
 gensync@^1.0.0-beta.1, gensync@^1.0.0-beta.2:
   version "1.0.0-beta.2"
   resolved "https://registry.yarnpkg.com/gensync/-/gensync-1.0.0-beta.2.tgz#32a6ee76c3d7f52d46b2b1ae5d93fea8580a25e0"
@@ -24101,6 +24140,18 @@ redis@^4.3.1:
     "@redis/search" "1.1.0"
     "@redis/time-series" "1.0.3"
 
+redis@^4.6.6:
+  version "4.6.6"
+  resolved "https://registry.yarnpkg.com/redis/-/redis-4.6.6.tgz#46d4f2d149d1634d6ef53db5747412a0ef7974ec"
+  integrity sha512-aLs2fuBFV/VJ28oLBqYykfnhGGkFxvx0HdCEBYdJ99FFbSEMZ7c1nVKwR6ZRv+7bb7JnC0mmCzaqu8frgOYhpA==
+  dependencies:
+    "@redis/bloom" "1.2.0"
+    "@redis/client" "1.5.7"
+    "@redis/graph" "1.1.0"
+    "@redis/json" "1.0.4"
+    "@redis/search" "1.1.2"
+    "@redis/time-series" "1.0.4"
+
 reflect-metadata@^0.1.13:
   version "0.1.13"
   resolved "https://registry.yarnpkg.com/reflect-metadata/-/reflect-metadata-0.1.13.tgz#67ae3ca57c972a2aa1642b10fe363fe32d49dc08"

From 67014e68fd2b82258ad3c69ae9279c7ed4cb603b Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Thu, 25 May 2023 13:43:33 +0800
Subject: [PATCH 05/12] feat: send email notification once the import has
 finished

---
 packages/import-handler/src/csv.ts            | 29 +++++++++++
 packages/import-handler/src/index.ts          | 48 ++++++++++++++++---
 .../src/luaScripts/updateMetrics.lua          |  3 ++
 packages/import-handler/src/metrics.ts        | 30 ++++++++----
 packages/puppeteer-parse/index.js             |  6 +--
 5 files changed, 96 insertions(+), 20 deletions(-)

diff --git a/packages/import-handler/src/csv.ts b/packages/import-handler/src/csv.ts
index 140cf00f0..2ff059735 100644
--- a/packages/import-handler/src/csv.ts
+++ b/packages/import-handler/src/csv.ts
@@ -6,6 +6,7 @@
 import { parse } from '@fast-csv/parse'
 import { Stream } from 'stream'
 import { ImportContext } from '.'
+import { ImportStatus, updateMetrics } from './metrics'
 
 export const importCsv = async (ctx: ImportContext, stream: Stream) => {
   const parser = parse()
@@ -23,11 +24,39 @@ export const importCsv = async (ctx: ImportContext, stream: Stream) => {
               .map((l) => l.trim())
               .filter((l) => l !== '')
           : undefined
+
+      // update total counter
+      await updateMetrics(
+        ctx.redisClient,
+        ctx.userId,
+        ctx.taskId,
+        ImportStatus.TOTAL,
+        ctx.source
+      )
+
       await ctx.urlHandler(ctx, url, state, labels)
+
       ctx.countImported += 1
+      // update started counter
+      await updateMetrics(
+        ctx.redisClient,
+        ctx.userId,
+        ctx.taskId,
+        ImportStatus.STARTED,
+        ctx.source
+      )
     } catch (error) {
       console.log('invalid url', row, error)
+
       ctx.countFailed += 1
+      // update invalid counter
+      await updateMetrics(
+        ctx.redisClient,
+        ctx.userId,
+        ctx.taskId,
+        ImportStatus.INVALID,
+        ctx.source
+      )
     }
   }
 }
diff --git a/packages/import-handler/src/index.ts b/packages/import-handler/src/index.ts
index 89b469ad1..53d02e926 100644
--- a/packages/import-handler/src/index.ts
+++ b/packages/import-handler/src/index.ts
@@ -5,12 +5,18 @@ import axios from 'axios'
 import * as jwt from 'jsonwebtoken'
 import { Stream } from 'node:stream'
 import * as path from 'path'
+import { createClient } from 'redis'
 import { promisify } from 'util'
 import { v4 as uuid } from 'uuid'
 import { importCsv } from './csv'
 import { importMatterArchive } from './matterHistory'
+import { createMetrics } from './metrics'
+import { createRedisClient } from './redis'
 import { CONTENT_FETCH_URL, createCloudTask, emailUserUrl } from './task'
 
+// explicitly create the return type of RedisClient
+type RedisClient = ReturnType<typeof createClient>
+
 export enum ArticleSavingRequestStatus {
   Failed = 'FAILED',
   Processing = 'PROCESSING',
@@ -51,6 +57,9 @@ export type ImportContext = {
   countFailed: number
   urlHandler: UrlHandler
   contentHandler: ContentHandler
+  redisClient: RedisClient
+  taskId: string
+  source: string
 }
 
 type importHandlerFunc = (ctx: ImportContext, stream: Stream) => Promise<void>
@@ -118,14 +127,27 @@ const sendImportFailedEmail = async (userId: string) => {
   })
 }
 
-const sendImportCompletedEmail = async (
+export const sendImportStartedEmail = async (
   userId: string,
   urlsEnqueued: number,
   urlsFailed: number
 ) => {
   return createEmailCloudTask(userId, {
-    subject: 'Your Omnivore import has completed processing',
-    body: `${urlsEnqueued} URLs have been processed and should be available in your library. ${urlsFailed} URLs failed to be parsed.`,
+    subject: 'Your Omnivore import has started',
+    body: `We have started processing ${urlsEnqueued} URLs. ${urlsFailed} URLs are invalid.`,
+  })
+}
+
+export const sendImportCompletedEmail = async (
+  userId: string,
+  urlsImported: number,
+  urlsFailed: number
+) => {
+  return createEmailCloudTask(userId, {
+    subject: 'Your Omnivore import has finished',
+    body: `We have finished processing ${
+      urlsImported + urlsFailed
+    } URLs. ${urlsImported} URLs have been added to your library. ${urlsFailed} URLs failed to be parsed.`,
   })
 }
 
@@ -222,7 +244,7 @@ const contentHandler = async (
   return Promise.resolve()
 }
 
-const handleEvent = async (data: StorageEvent) => {
+const handleEvent = async (data: StorageEvent, redisClient: RedisClient) => {
   if (shouldHandle(data)) {
     const handler = handlerForFile(data.name)
     if (!handler) {
@@ -253,12 +275,18 @@ const handleEvent = async (data: StorageEvent) => {
       countFailed: 0,
       urlHandler,
       contentHandler,
+      redisClient,
+      taskId: data.name,
+      source: 'csv-importer',
     }
 
+    // create metrics in redis
+    await createMetrics(redisClient, ctx.userId, ctx.taskId, ctx.source)
+
     await handler(ctx, stream)
 
     if (ctx.countImported > 0) {
-      await sendImportCompletedEmail(userId, ctx.countImported, ctx.countFailed)
+      await sendImportStartedEmail(userId, ctx.countImported, ctx.countFailed)
     } else {
       await sendImportFailedEmail(userId)
     }
@@ -285,11 +313,19 @@ export const importHandler = Sentry.GCPFunction.wrapHttpFunction(
       const pubSubMessage = req.body.message.data as string
       const obj = getStorageEvent(pubSubMessage)
       if (obj) {
+        // create redis client
+        const redisClient = await createRedisClient(
+          process.env.REDIS_URL,
+          process.env.REDIS_CERT
+        )
         try {
-          await handleEvent(obj)
+          await handleEvent(obj, redisClient)
         } catch (err) {
           console.log('error handling event', { err, obj })
           throw err
+        } finally {
+          // close redis client
+          await redisClient.quit()
         }
       }
     } else {
diff --git a/packages/import-handler/src/luaScripts/updateMetrics.lua b/packages/import-handler/src/luaScripts/updateMetrics.lua
index 94c45e1ab..4e7b08f26 100644
--- a/packages/import-handler/src/luaScripts/updateMetrics.lua
+++ b/packages/import-handler/src/luaScripts/updateMetrics.lua
@@ -27,5 +27,8 @@ if (status == "imported" or status == "failed") then
     -- all the records have been processed
     -- update the metrics
     redis.call('HSET', key, 'end_time', timestamp, 'state', 1);
+    return 1
   end
 end
+
+return 0;
diff --git a/packages/import-handler/src/metrics.ts b/packages/import-handler/src/metrics.ts
index 7be5edff5..75086639d 100644
--- a/packages/import-handler/src/metrics.ts
+++ b/packages/import-handler/src/metrics.ts
@@ -1,10 +1,11 @@
 import { createClient } from 'redis'
+import { sendImportCompletedEmail } from '.'
 import { lua } from './redis'
 
 // explicitly create the return type of RedisClient
 type RedisClient = ReturnType<typeof createClient>
 
-enum ImportStatus {
+export enum ImportStatus {
   STARTED = 'started',
   INVALID = 'invalid',
   IMPORTED = 'imported',
@@ -23,24 +24,24 @@ interface ImportMetrics {
   imported: number
   failed: number
   total: number
-  importer: string
+  source: string
   state: ImportTaskState
   startTime: number
   endTime: number
 }
 
-export const startImport = async (
+export const createMetrics = async (
   redisClient: RedisClient,
   userId: string,
   taskId: string,
-  importer: string
+  source: string
 ) => {
   const key = `import:${userId}:${taskId}`
   try {
     // set multiple fields
     await redisClient.hSet(key, {
-      ['start_time']: Date.now(), // unix timestamp in seconds
-      ['importer']: importer,
+      ['start_time']: Date.now(),
+      ['source']: source,
       ['state']: ImportTaskState.STARTED,
     })
   } catch (error) {
@@ -52,16 +53,25 @@ export const updateMetrics = async (
   redisClient: RedisClient,
   userId: string,
   taskId: string,
-  status: ImportStatus
+  status: ImportStatus,
+  source: string
 ) => {
   const key = `import:${userId}:${taskId}`
 
   try {
     // use lua script to increment hash field
-    await redisClient.evalSha(lua.sha, {
+    const state = await redisClient.evalSha(lua.sha, {
       keys: [key],
-      arguments: [status, Date.now().toString()],
+      arguments: [status, Date.now().toString(), source],
     })
+
+    // if the task is finished, send email
+    if (state == ImportTaskState.FINISHED) {
+      const metrics = await getMetrics(redisClient, userId, taskId)
+      if (metrics) {
+        await sendImportCompletedEmail(userId, metrics.imported, metrics.failed)
+      }
+    }
   } catch (error) {
     console.error('Redis Error', error)
   }
@@ -83,7 +93,7 @@ export const getMetrics = async (
       imported: parseInt(metrics.imported, 10),
       failed: parseInt(metrics.failed, 10),
       total: parseInt(metrics.total, 10),
-      importer: metrics.importer,
+      source: metrics.source,
       state: parseInt(metrics.state, 10),
       startTime: parseInt(metrics.start_time, 10),
       endTime: parseInt(metrics.end_time, 10),
diff --git a/packages/puppeteer-parse/index.js b/packages/puppeteer-parse/index.js
index b25498790..405f0a219 100644
--- a/packages/puppeteer-parse/index.js
+++ b/packages/puppeteer-parse/index.js
@@ -224,7 +224,7 @@ const sendSavePageMutation = async (userId, input) => {
           }
     }`,
     variables: {
-      input,
+      input: Object.assign({}, input , { source: 'puppeteer-parse' }),
     },
   });
 
@@ -256,7 +256,7 @@ async function fetchContent(req, res) {
   const articleSavingRequestId = (req.query ? req.query.saveRequestId : undefined) || (req.body ? req.body.saveRequestId : undefined);
   const state = req.body.state
   const labels = req.body.labels
-  const source = req.body.source || 'puppeteer-parse';
+  const source = req.body.source || 'parseContent';
 
   let logRecord = {
     url,
@@ -343,7 +343,6 @@ async function fetchContent(req, res) {
         parseResult: readabilityResult,
         state,
         labels,
-        source,
       });
 
       logRecord.totalTime = Date.now() - functionStartTime;
@@ -380,7 +379,6 @@ async function fetchContent(req, res) {
       parseResult: readabilityResult,
       state,
       labels,
-      source,
     });
 
     logRecord.totalTime = Date.now() - functionStartTime;

From 5f465112bfa402e4f5fbbaee773b15c98a27ee4b Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Thu, 25 May 2023 13:56:30 +0800
Subject: [PATCH 06/12] feat: update matter importer metrics

---
 packages/import-handler/src/csv.ts           | 14 ++--
 packages/import-handler/src/index.ts         |  6 --
 packages/import-handler/src/matterHistory.ts | 67 +++++++++++++++++---
 packages/import-handler/src/metrics.ts       |  5 +-
 4 files changed, 68 insertions(+), 24 deletions(-)

diff --git a/packages/import-handler/src/csv.ts b/packages/import-handler/src/csv.ts
index 2ff059735..70bb8b955 100644
--- a/packages/import-handler/src/csv.ts
+++ b/packages/import-handler/src/csv.ts
@@ -6,9 +6,12 @@
 import { parse } from '@fast-csv/parse'
 import { Stream } from 'stream'
 import { ImportContext } from '.'
-import { ImportStatus, updateMetrics } from './metrics'
+import { createMetrics, ImportStatus, updateMetrics } from './metrics'
 
 export const importCsv = async (ctx: ImportContext, stream: Stream) => {
+  // create metrics in redis
+  await createMetrics(ctx.redisClient, ctx.userId, ctx.taskId, 'csv-importer')
+
   const parser = parse()
   stream.pipe(parser)
   for await (const row of parser) {
@@ -30,8 +33,7 @@ export const importCsv = async (ctx: ImportContext, stream: Stream) => {
         ctx.redisClient,
         ctx.userId,
         ctx.taskId,
-        ImportStatus.TOTAL,
-        ctx.source
+        ImportStatus.TOTAL
       )
 
       await ctx.urlHandler(ctx, url, state, labels)
@@ -42,8 +44,7 @@ export const importCsv = async (ctx: ImportContext, stream: Stream) => {
         ctx.redisClient,
         ctx.userId,
         ctx.taskId,
-        ImportStatus.STARTED,
-        ctx.source
+        ImportStatus.STARTED
       )
     } catch (error) {
       console.log('invalid url', row, error)
@@ -54,8 +55,7 @@ export const importCsv = async (ctx: ImportContext, stream: Stream) => {
         ctx.redisClient,
         ctx.userId,
         ctx.taskId,
-        ImportStatus.INVALID,
-        ctx.source
+        ImportStatus.INVALID
       )
     }
   }
diff --git a/packages/import-handler/src/index.ts b/packages/import-handler/src/index.ts
index 53d02e926..b20c51e8d 100644
--- a/packages/import-handler/src/index.ts
+++ b/packages/import-handler/src/index.ts
@@ -10,7 +10,6 @@ import { promisify } from 'util'
 import { v4 as uuid } from 'uuid'
 import { importCsv } from './csv'
 import { importMatterArchive } from './matterHistory'
-import { createMetrics } from './metrics'
 import { createRedisClient } from './redis'
 import { CONTENT_FETCH_URL, createCloudTask, emailUserUrl } from './task'
 
@@ -59,7 +58,6 @@ export type ImportContext = {
   contentHandler: ContentHandler
   redisClient: RedisClient
   taskId: string
-  source: string
 }
 
 type importHandlerFunc = (ctx: ImportContext, stream: Stream) => Promise<void>
@@ -277,12 +275,8 @@ const handleEvent = async (data: StorageEvent, redisClient: RedisClient) => {
       contentHandler,
       redisClient,
       taskId: data.name,
-      source: 'csv-importer',
     }
 
-    // create metrics in redis
-    await createMetrics(redisClient, ctx.userId, ctx.taskId, ctx.source)
-
     await handler(ctx, stream)
 
     if (ctx.countImported > 0) {
diff --git a/packages/import-handler/src/matterHistory.ts b/packages/import-handler/src/matterHistory.ts
index 6b626d9bc..378f9d51e 100644
--- a/packages/import-handler/src/matterHistory.ts
+++ b/packages/import-handler/src/matterHistory.ts
@@ -4,20 +4,19 @@
 /* eslint-disable @typescript-eslint/no-unsafe-argument */
 
 import { parse } from '@fast-csv/parse'
-import { Stream } from 'stream'
-import unzip from 'unzip-stream'
+import { Readability } from '@omnivore/readability'
+import crypto from 'crypto'
+import createDOMPurify, { SanitizeElementHookEvent } from 'dompurify'
 import fs from 'fs'
-import path from 'path'
 import * as fsExtra from 'fs-extra'
 import glob from 'glob'
-
 import { parseHTML } from 'linkedom'
-import { Readability } from '@omnivore/readability'
-import createDOMPurify, { SanitizeElementHookEvent } from 'dompurify'
-
+import path from 'path'
+import { Stream } from 'stream'
+import unzip from 'unzip-stream'
 import { encode } from 'urlsafe-base64'
-import crypto from 'crypto'
 import { ImportContext } from '.'
+import { createMetrics, ImportStatus, updateMetrics } from './metrics'
 
 export type UrlHandler = (url: URL) => Promise<void>
 
@@ -36,8 +35,22 @@ export const importMatterHistoryCsv = async (
   for await (const row of parser) {
     try {
       const url = new URL(row['URL'])
+      // update total counter
+      await updateMetrics(
+        ctx.redisClient,
+        ctx.userId,
+        ctx.taskId,
+        ImportStatus.TOTAL
+      )
       await ctx.urlHandler(ctx, url)
       ctx.countImported += 1
+      // update started counter
+      await updateMetrics(
+        ctx.redisClient,
+        ctx.userId,
+        ctx.taskId,
+        ImportStatus.STARTED
+      )
     } catch (error) {
       console.log('invalid url', row, error)
       ctx.countFailed += 1
@@ -204,6 +217,13 @@ const handleMatterHistoryRow = async (
 
   if (!url) {
     ctx.countFailed += 1
+    // update failed counter
+    await updateMetrics(
+      ctx.redisClient,
+      ctx.userId,
+      ctx.taskId,
+      ImportStatus.FAILED
+    )
     return
   }
 
@@ -232,6 +252,14 @@ export const importMatterArchive = async (
   const archiveDir = await unarchive(stream)
 
   try {
+    // create metrics in redis
+    await createMetrics(
+      ctx.redisClient,
+      ctx.userId,
+      ctx.taskId,
+      'matter-importer'
+    )
+
     const historyFile = path.join(archiveDir, '_matter_history.csv')
 
     const parser = parse({
@@ -243,11 +271,34 @@ export const importMatterArchive = async (
 
     for await (const row of parser) {
       try {
+        // update total metrics
+        await updateMetrics(
+          ctx.redisClient,
+          ctx.userId,
+          ctx.taskId,
+          ImportStatus.TOTAL
+        )
+
         await handleMatterHistoryRow(ctx, archiveDir, row)
+
         ctx.countImported += 1
+        // update started metrics
+        await updateMetrics(
+          ctx.redisClient,
+          ctx.userId,
+          ctx.taskId,
+          ImportStatus.STARTED
+        )
       } catch (error) {
         console.log('invalid url', row, error)
         ctx.countFailed += 1
+        // update failed metrics
+        await updateMetrics(
+          ctx.redisClient,
+          ctx.userId,
+          ctx.taskId,
+          ImportStatus.FAILED
+        )
       }
     }
   } catch (err) {
diff --git a/packages/import-handler/src/metrics.ts b/packages/import-handler/src/metrics.ts
index 75086639d..1fef1f7f8 100644
--- a/packages/import-handler/src/metrics.ts
+++ b/packages/import-handler/src/metrics.ts
@@ -53,8 +53,7 @@ export const updateMetrics = async (
   redisClient: RedisClient,
   userId: string,
   taskId: string,
-  status: ImportStatus,
-  source: string
+  status: ImportStatus
 ) => {
   const key = `import:${userId}:${taskId}`
 
@@ -62,7 +61,7 @@ export const updateMetrics = async (
     // use lua script to increment hash field
     const state = await redisClient.evalSha(lua.sha, {
       keys: [key],
-      arguments: [status, Date.now().toString(), source],
+      arguments: [status, Date.now().toString()],
     })
 
     // if the task is finished, send email

From fba8e0424d89ceb621b774bae99b3250a6d9d184 Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Thu, 25 May 2023 15:27:28 +0800
Subject: [PATCH 07/12] feat: add importer metrics collector

---
 packages/import-handler/src/index.ts | 50 ++++++++++++++++++++++++++++
 1 file changed, 50 insertions(+)

diff --git a/packages/import-handler/src/index.ts b/packages/import-handler/src/index.ts
index b20c51e8d..e0121f0a1 100644
--- a/packages/import-handler/src/index.ts
+++ b/packages/import-handler/src/index.ts
@@ -10,6 +10,7 @@ import { promisify } from 'util'
 import { v4 as uuid } from 'uuid'
 import { importCsv } from './csv'
 import { importMatterArchive } from './matterHistory'
+import { ImportStatus, updateMetrics } from './metrics'
 import { createRedisClient } from './redis'
 import { CONTENT_FETCH_URL, createCloudTask, emailUserUrl } from './task'
 
@@ -62,6 +63,15 @@ export type ImportContext = {
 
 type importHandlerFunc = (ctx: ImportContext, stream: Stream) => Promise<void>
 
+interface UpdateMetricsRequest {
+  taskId: string
+  status: ImportStatus
+}
+
+function isUpdateMetricsRequest(body: any): body is UpdateMetricsRequest {
+  return 'taskId' in body && 'status' in body
+}
+
 interface StorageEvent {
   name: string
   bucket: string
@@ -328,3 +338,43 @@ export const importHandler = Sentry.GCPFunction.wrapHttpFunction(
     res.send('ok')
   }
 )
+
+export const importMetricsCollector = Sentry.GCPFunction.wrapHttpFunction(
+  async (req, res) => {
+    if (!process.env.JWT_SECRET) {
+      console.error('JWT_SECRET not exists')
+      return res.status(500).send({ errorCodes: 'JWT_SECRET_NOT_EXISTS' })
+    }
+    const token = (req.query.token || req.headers.authorization) as string
+    if (!token) {
+      return res.status(401).send({ errorCode: 'INVALID_TOKEN' })
+    }
+
+    let userId: string
+
+    try {
+      const decoded = jwt.verify(token, process.env.JWT_SECRET) as {
+        uid: string
+      }
+      userId = decoded.uid
+    } catch (e) {
+      console.error('Authentication error:', e)
+      return res.status(401).send({ errorCode: 'UNAUTHENTICATED' })
+    }
+
+    const redisClient = await createRedisClient(
+      process.env.REDIS_URL,
+      process.env.REDIS_CERT
+    )
+    if (!isUpdateMetricsRequest(req.body)) {
+      console.log('Invalid request body')
+      return res.status(400).send('Bad Request')
+    }
+
+    // update metrics
+    await updateMetrics(redisClient, userId, req.body.taskId, req.body.status)
+    await redisClient.quit()
+
+    res.send('ok')
+  }
+)

From 813a90e3c956b8f80344e510ff89095e46ef8a93 Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Thu, 25 May 2023 15:47:13 +0800
Subject: [PATCH 08/12] feat: send importer status update

---
 packages/import-handler/src/index.ts |  7 ++++-
 packages/puppeteer-parse/index.js    | 40 ++++++++++++++++++++++++++--
 2 files changed, 44 insertions(+), 3 deletions(-)

diff --git a/packages/import-handler/src/index.ts b/packages/import-handler/src/index.ts
index e0121f0a1..28f6dd958 100644
--- a/packages/import-handler/src/index.ts
+++ b/packages/import-handler/src/index.ts
@@ -40,6 +40,7 @@ const CONTENT_TYPES = ['text/csv', 'application/zip']
 export type UrlHandler = (
   ctx: ImportContext,
   url: URL,
+  taskId: string,
   state?: ArticleSavingRequestStatus,
   labels?: string[]
 ) => Promise<void>
@@ -96,6 +97,7 @@ const importURL = async (
   userId: string,
   url: URL,
   source: string,
+  taskId: string,
   state?: ArticleSavingRequestStatus,
   labels?: string[]
 ): Promise<string | undefined> => {
@@ -173,6 +175,7 @@ const handlerForFile = (name: string): importHandlerFunc | undefined => {
 const urlHandler = async (
   ctx: ImportContext,
   url: URL,
+  taskId: string,
   state?: ArticleSavingRequestStatus,
   labels?: string[]
 ): Promise<void> => {
@@ -182,6 +185,7 @@ const urlHandler = async (
       ctx.userId,
       url,
       'csv-importer',
+      taskId,
       state,
       labels && labels.length > 0 ? labels : undefined
     )
@@ -345,7 +349,7 @@ export const importMetricsCollector = Sentry.GCPFunction.wrapHttpFunction(
       console.error('JWT_SECRET not exists')
       return res.status(500).send({ errorCodes: 'JWT_SECRET_NOT_EXISTS' })
     }
-    const token = (req.query.token || req.headers.authorization) as string
+    const token = req.headers.authorization
     if (!token) {
       return res.status(401).send({ errorCode: 'INVALID_TOKEN' })
     }
@@ -373,6 +377,7 @@ export const importMetricsCollector = Sentry.GCPFunction.wrapHttpFunction(
 
     // update metrics
     await updateMetrics(redisClient, userId, req.body.taskId, req.body.status)
+
     await redisClient.quit()
 
     res.send('ok')
diff --git a/packages/puppeteer-parse/index.js b/packages/puppeteer-parse/index.js
index 405f0a219..09fc38e9e 100644
--- a/packages/puppeteer-parse/index.js
+++ b/packages/puppeteer-parse/index.js
@@ -44,6 +44,8 @@ const NON_SCRIPT_HOSTS= ['medium.com', 'fastcompany.com'];
 
 const ALLOWED_CONTENT_TYPES = ['text/html', 'application/octet-stream', 'text/plain', 'application/pdf'];
 
+const IMPORTER_METRICS_COLLECTOR_URL = process.env.IMPORTER_METRICS_COLLECTOR_URL;
+
 const userAgentForUrl = (url) => {
   try {
     const u = new URL(url);
@@ -248,6 +250,25 @@ const saveUploadedPdf = async (userId, url, uploadFileId, articleSavingRequestId
   );
 };
 
+const sendImportStatusUpdate = async (userId, taskId, status) => {
+  const auth = await signToken({ uid: userId }, process.env.JWT_SECRET);
+
+  const response = await axios.post(
+    IMPORTER_METRICS_COLLECTOR_URL, 
+    {
+      taskId,
+      status,
+    },
+    {
+      headers: {
+        'Authorization': auth,
+        'Content-Type': 'application/json',
+      },
+    });
+
+  return response.data;
+};
+
 async function fetchContent(req, res) {
   let functionStartTime = Date.now();
 
@@ -257,6 +278,7 @@ async function fetchContent(req, res) {
   const state = req.body.state
   const labels = req.body.labels
   const source = req.body.source || 'parseContent';
+  const taskId = req.body.taskId; // taskId is used to update import status
 
   let logRecord = {
     url,
@@ -266,7 +288,8 @@ async function fetchContent(req, res) {
       source,
     },
     state,
-    labelsToAdd: labels
+    labelsToAdd: labels,
+    taskId: taskId,
   };
 
   console.info(`Article parsing request`, logRecord);
@@ -278,7 +301,7 @@ async function fetchContent(req, res) {
   }
 
   // pre handle url with custom handlers
-  let title, content, contentType;
+  let title, content, contentType, importStatus;
   try {
     const browser = await getBrowserPromise;
     const result = await preHandleContent(url, browser);
@@ -348,6 +371,8 @@ async function fetchContent(req, res) {
       logRecord.totalTime = Date.now() - functionStartTime;
       logRecord.result = apiResponse.createArticle;
     }
+
+    importStatus = 'imported';
   } catch (e) {
     logRecord.error = e.message;
     console.error(`Error while retrieving page`, logRecord);
@@ -383,12 +408,23 @@ async function fetchContent(req, res) {
 
     logRecord.totalTime = Date.now() - functionStartTime;
     logRecord.result = apiResponse.createArticle;
+
+    importStatus = 'failed';
   } finally {
     if (context) {
       await context.close();
     }
     console.info(`parse-page`, logRecord);
 
+    // send import status to update the metrics
+    if (taskId) {
+      try {
+        await sendImportStatusUpdate(userId, taskId, importStatus);
+      } catch (e) {
+        console.error('Error while sending import status update', e);
+      }
+    }
+
     res.sendStatus(200);
   }
 }

From 1073f88d69f9c281bd73e13eeddabd3691b8985c Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Thu, 25 May 2023 15:51:48 +0800
Subject: [PATCH 09/12] fix: add start:collector command

---
 packages/import-handler/package.json | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/packages/import-handler/package.json b/packages/import-handler/package.json
index d13ce1623..d1e6f2b34 100644
--- a/packages/import-handler/package.json
+++ b/packages/import-handler/package.json
@@ -15,7 +15,8 @@
     "build": "tsc && yarn copy-files",
     "start": "functions-framework --target=importHandler",
     "dev": "concurrently \"tsc -w\" \"nodemon --watch ./build/ --exec npm run start\"",
-    "copy-files": "copyfiles -u 1 src/**/*.lua build/src"
+    "copy-files": "copyfiles -u 1 src/**/*.lua build/src",
+    "start:collector": "functions-framework --target=importMetricsCollector"
   },
   "devDependencies": {
     "@types/chai": "^4.3.4",

From 345c56c57a40857011fc23cdb3859674d4d35b34 Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Thu, 25 May 2023 15:53:39 +0800
Subject: [PATCH 10/12] fix: catch importer status update error

---
 packages/puppeteer-parse/index.js | 36 +++++++++++++++----------------
 1 file changed, 17 insertions(+), 19 deletions(-)

diff --git a/packages/puppeteer-parse/index.js b/packages/puppeteer-parse/index.js
index 09fc38e9e..7c5bda58a 100644
--- a/packages/puppeteer-parse/index.js
+++ b/packages/puppeteer-parse/index.js
@@ -251,22 +251,24 @@ const saveUploadedPdf = async (userId, url, uploadFileId, articleSavingRequestId
 };
 
 const sendImportStatusUpdate = async (userId, taskId, status) => {
-  const auth = await signToken({ uid: userId }, process.env.JWT_SECRET);
+  try {
+    const auth = await signToken({ uid: userId }, process.env.JWT_SECRET);
 
-  const response = await axios.post(
-    IMPORTER_METRICS_COLLECTOR_URL, 
-    {
-      taskId,
-      status,
-    },
-    {
-      headers: {
-        'Authorization': auth,
-        'Content-Type': 'application/json',
+    await axios.post(
+      IMPORTER_METRICS_COLLECTOR_URL, 
+      {
+        taskId,
+        status,
       },
-    });
-
-  return response.data;
+      {
+        headers: {
+          'Authorization': auth,
+          'Content-Type': 'application/json',
+        },
+      });
+  } catch (e) {
+    console.error('Error while sending import status update', e);
+  }
 };
 
 async function fetchContent(req, res) {
@@ -418,11 +420,7 @@ async function fetchContent(req, res) {
 
     // send import status to update the metrics
     if (taskId) {
-      try {
-        await sendImportStatusUpdate(userId, taskId, importStatus);
-      } catch (e) {
-        console.error('Error while sending import status update', e);
-      }
+      await sendImportStatusUpdate(userId, taskId, importStatus);
     }
 
     res.sendStatus(200);

From 8a89a47c166b5d49c6056bd8e0ae3a5fded31243 Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Thu, 25 May 2023 16:08:43 +0800
Subject: [PATCH 11/12] fix: testcases

---
 .github/workflows/run-tests.yaml                   |  9 +++++++++
 packages/import-handler/src/index.ts               |  4 +---
 packages/import-handler/src/redis.ts               |  8 ++++++--
 packages/import-handler/test/csv/csv.test.ts       | 12 +++++++++---
 .../test/matter/matter_importer.test.ts            | 14 +++++++++-----
 packages/import-handler/test/util.ts               |  6 +++++-
 6 files changed, 39 insertions(+), 14 deletions(-)

diff --git a/.github/workflows/run-tests.yaml b/.github/workflows/run-tests.yaml
index 0086a086e..e45c0f6c3 100644
--- a/.github/workflows/run-tests.yaml
+++ b/.github/workflows/run-tests.yaml
@@ -42,6 +42,15 @@ jobs:
           --health-retries 10
         ports:
           - 9200
+      redis:
+        image: redis
+        options: >-
+          --health-cmd "redis-cli ping"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 10
+        ports:
+          - 6379
     steps:
       - uses: actions/checkout@v2
         with:
diff --git a/packages/import-handler/src/index.ts b/packages/import-handler/src/index.ts
index 28f6dd958..b90537a38 100644
--- a/packages/import-handler/src/index.ts
+++ b/packages/import-handler/src/index.ts
@@ -40,7 +40,6 @@ const CONTENT_TYPES = ['text/csv', 'application/zip']
 export type UrlHandler = (
   ctx: ImportContext,
   url: URL,
-  taskId: string,
   state?: ArticleSavingRequestStatus,
   labels?: string[]
 ) => Promise<void>
@@ -175,7 +174,6 @@ const handlerForFile = (name: string): importHandlerFunc | undefined => {
 const urlHandler = async (
   ctx: ImportContext,
   url: URL,
-  taskId: string,
   state?: ArticleSavingRequestStatus,
   labels?: string[]
 ): Promise<void> => {
@@ -185,7 +183,7 @@ const urlHandler = async (
       ctx.userId,
       url,
       'csv-importer',
-      taskId,
+      ctx.taskId,
       state,
       labels && labels.length > 0 ? labels : undefined
     )
diff --git a/packages/import-handler/src/redis.ts b/packages/import-handler/src/redis.ts
index 3804c36ed..481f136b9 100644
--- a/packages/import-handler/src/redis.ts
+++ b/packages/import-handler/src/redis.ts
@@ -1,9 +1,13 @@
-import fs from 'fs'
+import { readFileSync } from 'fs'
+import path from 'path'
 import { createClient } from 'redis'
 
 // load lua script
 export const lua = {
-  script: fs.readFileSync('./luaScripts/updateMetrics.lua', 'utf8'),
+  script: readFileSync(
+    path.resolve(__dirname, 'luaScripts/updateMetrics.lua'),
+    'utf8'
+  ),
   sha: '',
 }
 
diff --git a/packages/import-handler/test/csv/csv.test.ts b/packages/import-handler/test/csv/csv.test.ts
index f045a0c96..21d4e175a 100644
--- a/packages/import-handler/test/csv/csv.test.ts
+++ b/packages/import-handler/test/csv/csv.test.ts
@@ -13,7 +13,7 @@ describe('Load a simple CSV file', () => {
   it('should call the handler for each URL', async () => {
     const urls: URL[] = []
     const stream = fs.createReadStream('./test/csv/data/simple.csv')
-    const stub = stubImportCtx()
+    const stub = await stubImportCtx()
     stub.urlHandler = (ctx: ImportContext, url): Promise<void> => {
       urls.push(url)
       return Promise.resolve()
@@ -26,12 +26,14 @@ describe('Load a simple CSV file', () => {
       new URL('https://omnivore.app'),
       new URL('https://google.com'),
     ])
+
+    await stub.redisClient.quit()
   })
 
   it('increments the failed count when the URL is invalid', async () => {
     const urls: URL[] = []
     const stream = fs.createReadStream('./test/csv/data/simple.csv')
-    const stub = stubImportCtx()
+    const stub = await stubImportCtx()
     stub.urlHandler = (ctx: ImportContext, url): Promise<void> => {
       urls.push(url)
       return Promise.reject('Failed to import url')
@@ -40,6 +42,8 @@ describe('Load a simple CSV file', () => {
     await importCsv(stub, stream)
     expect(stub.countFailed).to.equal(2)
     expect(stub.countImported).to.equal(0)
+
+    await stub.redisClient.quit()
   })
 })
 
@@ -51,7 +55,7 @@ describe('Load a complex CSV file', () => {
       labels?: string[]
     }[] = []
     const stream = fs.createReadStream('./test/csv/data/complex.csv')
-    const stub = stubImportCtx()
+    const stub = await stubImportCtx()
     stub.urlHandler = (
       ctx: ImportContext,
       url,
@@ -86,6 +90,8 @@ describe('Load a complex CSV file', () => {
         labels: ['test', 'development'],
       },
     ])
+
+    await stub.redisClient.quit()
   })
 })
 
diff --git a/packages/import-handler/test/matter/matter_importer.test.ts b/packages/import-handler/test/matter/matter_importer.test.ts
index 90600b16a..94017e634 100644
--- a/packages/import-handler/test/matter/matter_importer.test.ts
+++ b/packages/import-handler/test/matter/matter_importer.test.ts
@@ -1,15 +1,15 @@
-import 'mocha'
+import { Readability } from '@omnivore/readability'
 import * as chai from 'chai'
 import { expect } from 'chai'
 import chaiString from 'chai-string'
 import * as fs from 'fs'
+import 'mocha'
+import { ImportContext } from '../../src'
 import {
   importMatterArchive,
   importMatterHistoryCsv,
 } from '../../src/matterHistory'
 import { stubImportCtx } from '../util'
-import { ImportContext } from '../../src'
-import { Readability } from '@omnivore/readability'
 
 chai.use(chaiString)
 
@@ -17,7 +17,7 @@ describe('Load a simple _matter_history file', () => {
   it('should find the URL of each row', async () => {
     const urls: URL[] = []
     const stream = fs.createReadStream('./test/matter/data/_matter_history.csv')
-    const stub = stubImportCtx()
+    const stub = await stubImportCtx()
     stub.urlHandler = (ctx: ImportContext, url): Promise<void> => {
       urls.push(url)
       return Promise.resolve()
@@ -29,6 +29,8 @@ describe('Load a simple _matter_history file', () => {
     expect(urls).to.eql([
       new URL('https://www.bloomberg.com/features/2022-the-crypto-story/'),
     ])
+
+    await stub.redisClient.quit()
   })
 })
 
@@ -36,7 +38,7 @@ describe('Load archive file', () => {
   it('should find the URL of each row', async () => {
     const urls: URL[] = []
     const stream = fs.createReadStream('./test/matter/data/Archive.zip')
-    const stub = stubImportCtx()
+    const stub = await stubImportCtx()
     stub.contentHandler = (
       ctx: ImportContext,
       url: URL,
@@ -54,5 +56,7 @@ describe('Load archive file', () => {
     expect(urls).to.eql([
       new URL('https://www.bloomberg.com/features/2022-the-crypto-story/'),
     ])
+
+    await stub.redisClient.quit()
   })
 })
diff --git a/packages/import-handler/test/util.ts b/packages/import-handler/test/util.ts
index a60564ae2..5fa9ba785 100644
--- a/packages/import-handler/test/util.ts
+++ b/packages/import-handler/test/util.ts
@@ -1,7 +1,9 @@
 import { Readability } from '@omnivore/readability'
 import { ArticleSavingRequestStatus, ImportContext } from '../src'
+import { createRedisClient } from '../src/redis'
 
-export const stubImportCtx = () => {
+export const stubImportCtx = async () => {
+  const redisClient = await createRedisClient()
   return {
     userId: '',
     countImported: 0,
@@ -23,5 +25,7 @@ export const stubImportCtx = () => {
     ): Promise<void> => {
       return Promise.resolve()
     },
+    redisClient,
+    taskId: '',
   }
 }

From 992a78e17338c25562a02093493580adbf73e6ac Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Thu, 25 May 2023 16:14:40 +0800
Subject: [PATCH 12/12] add dockerfile

---
 .github/workflows/run-tests.yaml             |   1 +
 packages/import-handler/Dockerfile-collector |  31 +++
 packages/import-handler/package.json         |   4 +-
 packages/import-handler/src/index.ts         |   8 +-
 packages/import-handler/test/csv/csv.test.ts | 233 +++++++++----------
 packages/import-handler/test/util.ts         |   5 +-
 packages/import-handler/tsconfig.json        |   4 +-
 yarn.lock                                    |  51 ----
 8 files changed, 159 insertions(+), 178 deletions(-)
 create mode 100644 packages/import-handler/Dockerfile-collector

diff --git a/.github/workflows/run-tests.yaml b/.github/workflows/run-tests.yaml
index e45c0f6c3..08bc418cb 100644
--- a/.github/workflows/run-tests.yaml
+++ b/.github/workflows/run-tests.yaml
@@ -98,6 +98,7 @@ jobs:
           PG_DB: omnivore_test
           PG_POOL_MAX: 10
           ELASTIC_URL: http://localhost:${{ job.services.elastic.ports[9200] }}/
+          REDIS_URL: redis://localhost:${{ job.services.redis.ports[6379] }}
   build-docker-images:
     name: Build docker images
     runs-on: ubuntu-latest
diff --git a/packages/import-handler/Dockerfile-collector b/packages/import-handler/Dockerfile-collector
new file mode 100644
index 000000000..fba468482
--- /dev/null
+++ b/packages/import-handler/Dockerfile-collector
@@ -0,0 +1,31 @@
+FROM node:14.18-alpine
+
+WORKDIR /app
+
+ENV PUPPETEER_SKIP_CHROMIUM_DOWNLOAD true
+RUN apk add g++ make python3
+
+ENV PORT 8080
+
+COPY package.json .
+COPY yarn.lock .
+COPY tsconfig.json .
+COPY .eslintrc .
+
+COPY /packages/readabilityjs/package.json ./packages/readabilityjs/package.json
+COPY /packages/import-handler/package.json ./packages/import-handler/package.json
+
+RUN yarn install --pure-lockfile
+
+ADD /packages/import-handler ./packages/import-handler
+ADD /packages/readabilityjs ./packages/readabilityjs
+RUN yarn workspace @omnivore/import-handler build
+
+# After building, fetch the production dependencies
+RUN rm -rf /app/packages/import-handler/node_modules
+RUN rm -rf /app/node_modules
+RUN yarn install --pure-lockfile --production
+
+EXPOSE 8080
+
+ENTRYPOINT ["yarn", "workspace", "@omnivore/import-handler", "start:collector"]
diff --git a/packages/import-handler/package.json b/packages/import-handler/package.json
index d1e6f2b34..ec14ef577 100644
--- a/packages/import-handler/package.json
+++ b/packages/import-handler/package.json
@@ -15,7 +15,7 @@
     "build": "tsc && yarn copy-files",
     "start": "functions-framework --target=importHandler",
     "dev": "concurrently \"tsc -w\" \"nodemon --watch ./build/ --exec npm run start\"",
-    "copy-files": "copyfiles -u 1 src/**/*.lua build/src",
+    "copy-files": "copyfiles src/luaScripts/*.lua build/",
     "start:collector": "functions-framework --target=importMetricsCollector"
   },
   "devDependencies": {
@@ -49,7 +49,7 @@
     "jsonwebtoken": "^8.5.1",
     "linkedom": "^0.14.21",
     "nodemon": "^2.0.15",
-    "redis": "^4.6.6",
+    "redis": "^4.3.1",
     "unzip-stream": "^0.3.1",
     "urlsafe-base64": "^1.0.0",
     "uuid": "^9.0.0"
diff --git a/packages/import-handler/src/index.ts b/packages/import-handler/src/index.ts
index b90537a38..269ad463c 100644
--- a/packages/import-handler/src/index.ts
+++ b/packages/import-handler/src/index.ts
@@ -364,15 +364,15 @@ export const importMetricsCollector = Sentry.GCPFunction.wrapHttpFunction(
       return res.status(401).send({ errorCode: 'UNAUTHENTICATED' })
     }
 
-    const redisClient = await createRedisClient(
-      process.env.REDIS_URL,
-      process.env.REDIS_CERT
-    )
     if (!isUpdateMetricsRequest(req.body)) {
       console.log('Invalid request body')
       return res.status(400).send('Bad Request')
     }
 
+    const redisClient = await createRedisClient(
+      process.env.REDIS_URL,
+      process.env.REDIS_CERT
+    )
     // update metrics
     await updateMetrics(redisClient, userId, req.body.taskId, req.body.status)
 
diff --git a/packages/import-handler/test/csv/csv.test.ts b/packages/import-handler/test/csv/csv.test.ts
index 21d4e175a..82798fa89 100644
--- a/packages/import-handler/test/csv/csv.test.ts
+++ b/packages/import-handler/test/csv/csv.test.ts
@@ -9,135 +9,134 @@ import { stubImportCtx } from '../util'
 
 chai.use(chaiString)
 
-describe('Load a simple CSV file', () => {
-  it('should call the handler for each URL', async () => {
-    const urls: URL[] = []
-    const stream = fs.createReadStream('./test/csv/data/simple.csv')
-    const stub = await stubImportCtx()
-    stub.urlHandler = (ctx: ImportContext, url): Promise<void> => {
-      urls.push(url)
-      return Promise.resolve()
-    }
+describe('Test csv importer', () => {
+  let stub: ImportContext
 
-    await importCsv(stub, stream)
-    expect(stub.countFailed).to.equal(0)
-    expect(stub.countImported).to.equal(2)
-    expect(urls).to.eql([
-      new URL('https://omnivore.app'),
-      new URL('https://google.com'),
-    ])
+  beforeEach(async () => {
+    stub = await stubImportCtx()
+  })
 
+  afterEach(async () => {
     await stub.redisClient.quit()
   })
 
-  it('increments the failed count when the URL is invalid', async () => {
-    const urls: URL[] = []
-    const stream = fs.createReadStream('./test/csv/data/simple.csv')
-    const stub = await stubImportCtx()
-    stub.urlHandler = (ctx: ImportContext, url): Promise<void> => {
-      urls.push(url)
-      return Promise.reject('Failed to import url')
-    }
+  describe('Load a simple CSV file', () => {
+    it('should call the handler for each URL', async () => {
+      const urls: URL[] = []
+      const stream = fs.createReadStream('./test/csv/data/simple.csv')
+      stub.urlHandler = (ctx: ImportContext, url): Promise<void> => {
+        urls.push(url)
+        return Promise.resolve()
+      }
 
-    await importCsv(stub, stream)
-    expect(stub.countFailed).to.equal(2)
-    expect(stub.countImported).to.equal(0)
+      await importCsv(stub, stream)
+      expect(stub.countFailed).to.equal(0)
+      expect(stub.countImported).to.equal(2)
+      expect(urls).to.eql([
+        new URL('https://omnivore.app'),
+        new URL('https://google.com'),
+      ])
+    })
 
-    await stub.redisClient.quit()
+    it('increments the failed count when the URL is invalid', async () => {
+      const stream = fs.createReadStream('./test/csv/data/simple.csv')
+      stub.urlHandler = (ctx: ImportContext, url): Promise<void> => {
+        return Promise.reject('Failed to import url')
+      }
+
+      await importCsv(stub, stream)
+      expect(stub.countFailed).to.equal(2)
+      expect(stub.countImported).to.equal(0)
+    })
   })
-})
 
-describe('Load a complex CSV file', () => {
-  it('should call the handler for each URL, state and labels', async () => {
-    const results: {
-      url: URL
-      state?: ArticleSavingRequestStatus
-      labels?: string[]
-    }[] = []
-    const stream = fs.createReadStream('./test/csv/data/complex.csv')
-    const stub = await stubImportCtx()
-    stub.urlHandler = (
-      ctx: ImportContext,
-      url,
-      state,
-      labels
-    ): Promise<void> => {
-      results.push({
+  describe('Load a complex CSV file', () => {
+    it('should call the handler for each URL, state and labels', async () => {
+      const results: {
+        url: URL
+        state?: ArticleSavingRequestStatus
+        labels?: string[]
+      }[] = []
+      const stream = fs.createReadStream('./test/csv/data/complex.csv')
+      stub.urlHandler = (
+        ctx: ImportContext,
         url,
         state,
-        labels,
-      })
-      return Promise.resolve()
-    }
+        labels
+      ): Promise<void> => {
+        results.push({
+          url,
+          state,
+          labels,
+        })
+        return Promise.resolve()
+      }
 
-    await importCsv(stub, stream)
-    expect(stub.countFailed).to.equal(0)
-    expect(stub.countImported).to.equal(3)
-    expect(results).to.eql([
-      {
-        url: new URL('https://omnivore.app'),
-        state: 'ARCHIVED',
-        labels: ['test'],
-      },
-      {
-        url: new URL('https://google.com'),
-        state: 'SUCCEEDED',
-        labels: ['test', 'development'],
-      },
-      {
-        url: new URL('https://test.com'),
-        state: 'SUCCEEDED',
-        labels: ['test', 'development'],
-      },
-    ])
+      await importCsv(stub, stream)
+      expect(stub.countFailed).to.equal(0)
+      expect(stub.countImported).to.equal(3)
+      expect(results).to.eql([
+        {
+          url: new URL('https://omnivore.app'),
+          state: 'ARCHIVED',
+          labels: ['test'],
+        },
+        {
+          url: new URL('https://google.com'),
+          state: 'SUCCEEDED',
+          labels: ['test', 'development'],
+        },
+        {
+          url: new URL('https://test.com'),
+          state: 'SUCCEEDED',
+          labels: ['test', 'development'],
+        },
+      ])
+    })
+  })
 
-    await stub.redisClient.quit()
-  })
-})
-
-describe('A file with no status set', () => {
-  it('should not try to set status', async () => {
-    const states: (ArticleSavingRequestStatus | undefined)[] = []
-    const stream = fs.createReadStream('./test/csv/data/unset-status.csv')
-    const stub = stubImportCtx()
-    stub.urlHandler = (
-      ctx: ImportContext,
-      url,
-      state?: ArticleSavingRequestStatus
-    ): Promise<void> => {
-      states.push(state)
-      return Promise.resolve()
-    }
-
-    await importCsv(stub, stream)
-    expect(stub.countFailed).to.equal(0)
-    expect(stub.countImported).to.equal(2)
-    expect(states).to.eql([undefined, ArticleSavingRequestStatus.Archived])
-  })
-})
-
-describe('A file with some labels', () => {
-  it('gets the labels, handles empty, and trims extra whitespace', async () => {
-    const importedLabels: (string[] | undefined)[] = []
-    const stream = fs.createReadStream('./test/csv/data/labels.csv')
-    const stub = stubImportCtx()
-    stub.urlHandler = (
-      ctx: ImportContext,
-      url,
-      state?: ArticleSavingRequestStatus,
-      labels?: string[]
-    ): Promise<void> => {
-      importedLabels.push(labels)
-      return Promise.resolve()
-    }
-
-    await importCsv(stub, stream)
-    expect(stub.countFailed).to.equal(0)
-    expect(stub.countImported).to.equal(3)
-    expect(importedLabels).to.eql([
-      ['Label1', 'Label2', 'Label 3', 'Label 4'],
-      [],
-      [],
-    ])
+  describe('A file with no status set', () => {
+    it('should not try to set status', async () => {
+      const states: (ArticleSavingRequestStatus | undefined)[] = []
+      const stream = fs.createReadStream('./test/csv/data/unset-status.csv')
+      stub.urlHandler = (
+        ctx: ImportContext,
+        url,
+        state?: ArticleSavingRequestStatus
+      ): Promise<void> => {
+        states.push(state)
+        return Promise.resolve()
+      }
+
+      await importCsv(stub, stream)
+      expect(stub.countFailed).to.equal(0)
+      expect(stub.countImported).to.equal(2)
+      expect(states).to.eql([undefined, ArticleSavingRequestStatus.Archived])
+    })
+  })
+
+  describe('A file with some labels', () => {
+    it('gets the labels, handles empty, and trims extra whitespace', async () => {
+      const importedLabels: (string[] | undefined)[] = []
+      const stream = fs.createReadStream('./test/csv/data/labels.csv')
+      stub.urlHandler = (
+        ctx: ImportContext,
+        url,
+        state?: ArticleSavingRequestStatus,
+        labels?: string[]
+      ): Promise<void> => {
+        importedLabels.push(labels)
+        return Promise.resolve()
+      }
+
+      await importCsv(stub, stream)
+      expect(stub.countFailed).to.equal(0)
+      expect(stub.countImported).to.equal(3)
+      expect(importedLabels).to.eql([
+        ['Label1', 'Label2', 'Label 3', 'Label 4'],
+        [],
+        [],
+      ])
+    })
   })
 })
diff --git a/packages/import-handler/test/util.ts b/packages/import-handler/test/util.ts
index 5fa9ba785..54a8b5082 100644
--- a/packages/import-handler/test/util.ts
+++ b/packages/import-handler/test/util.ts
@@ -2,8 +2,9 @@ import { Readability } from '@omnivore/readability'
 import { ArticleSavingRequestStatus, ImportContext } from '../src'
 import { createRedisClient } from '../src/redis'
 
-export const stubImportCtx = async () => {
-  const redisClient = await createRedisClient()
+export const stubImportCtx = async (): Promise<ImportContext> => {
+  const redisClient = await createRedisClient(process.env.REDIS_URL)
+
   return {
     userId: '',
     countImported: 0,
diff --git a/packages/import-handler/tsconfig.json b/packages/import-handler/tsconfig.json
index ea8c4d3ef..912ebd323 100644
--- a/packages/import-handler/tsconfig.json
+++ b/packages/import-handler/tsconfig.json
@@ -1,6 +1,6 @@
 {
   "extends": "./../../tsconfig.json",
-  "ts-node": {  "files": true },
+  "ts-node": { "files": true },
   "compilerOptions": {
     "outDir": "build",
     "rootDir": ".",
@@ -8,5 +8,5 @@
     // Generate d.ts files
     "declaration": true
   },
-  "include": ["src", "test"]
+  "include": ["src/**/*", "test/**/*"]
 }
diff --git a/yarn.lock b/yarn.lock
index 5ef541069..339090cfe 100644
--- a/yarn.lock
+++ b/yarn.lock
@@ -5431,11 +5431,6 @@
   resolved "https://registry.yarnpkg.com/@redis/bloom/-/bloom-1.0.2.tgz#42b82ec399a92db05e29fffcdfd9235a5fc15cdf"
   integrity sha512-EBw7Ag1hPgFzdznK2PBblc1kdlj5B5Cw3XwI9/oG7tSn85/HKy3X9xHy/8tm/eNXJYHLXHJL/pkwBpFMVVefkw==
 
-"@redis/bloom@1.2.0":
-  version "1.2.0"
-  resolved "https://registry.yarnpkg.com/@redis/bloom/-/bloom-1.2.0.tgz#d3fd6d3c0af3ef92f26767b56414a370c7b63b71"
-  integrity sha512-HG2DFjYKbpNmVXsa0keLHp/3leGJz1mjh09f2RLGGLQZzSHpkmZWuwJbAvo3QcRY8p80m5+ZdXZdYOSBLlp7Cg==
-
 "@redis/client@1.3.0":
   version "1.3.0"
   resolved "https://registry.yarnpkg.com/@redis/client/-/client-1.3.0.tgz#c62ccd707f16370a2dc2f9e158a28b7da049fa77"
@@ -5445,25 +5440,11 @@
     generic-pool "3.8.2"
     yallist "4.0.0"
 
-"@redis/client@1.5.7":
-  version "1.5.7"
-  resolved "https://registry.yarnpkg.com/@redis/client/-/client-1.5.7.tgz#92cc5c98c76f189e37d24f0e1e17e104c6af17d4"
-  integrity sha512-gaOBOuJPjK5fGtxSseaKgSvjiZXQCdLlGg9WYQst+/GRUjmXaiB5kVkeQMRtPc7Q2t93XZcJfBMSwzs/XS9UZw==
-  dependencies:
-    cluster-key-slot "1.1.2"
-    generic-pool "3.9.0"
-    yallist "4.0.0"
-
 "@redis/graph@1.0.1":
   version "1.0.1"
   resolved "https://registry.yarnpkg.com/@redis/graph/-/graph-1.0.1.tgz#eabc58ba99cd70d0c907169c02b55497e4ec8a99"
   integrity sha512-oDE4myMCJOCVKYMygEMWuriBgqlS5FqdWerikMoJxzmmTUErnTRRgmIDa2VcgytACZMFqpAOWDzops4DOlnkfQ==
 
-"@redis/graph@1.1.0":
-  version "1.1.0"
-  resolved "https://registry.yarnpkg.com/@redis/graph/-/graph-1.1.0.tgz#cc2b82e5141a29ada2cce7d267a6b74baa6dd519"
-  integrity sha512-16yZWngxyXPd+MJxeSr0dqh2AIOi8j9yXKcKCwVaKDbH3HTuETpDVPcLujhFYVPtYrngSco31BUcSa9TH31Gqg==
-
 "@redis/json@1.0.4":
   version "1.0.4"
   resolved "https://registry.yarnpkg.com/@redis/json/-/json-1.0.4.tgz#f372b5f93324e6ffb7f16aadcbcb4e5c3d39bda1"
@@ -5474,21 +5455,11 @@
   resolved "https://registry.yarnpkg.com/@redis/search/-/search-1.1.0.tgz#7abb18d431f27ceafe6bcb4dd83a3fa67e9ab4df"
   integrity sha512-NyFZEVnxIJEybpy+YskjgOJRNsfTYqaPbK/Buv6W2kmFNaRk85JiqjJZA5QkRmWvGbyQYwoO5QfDi2wHskKrQQ==
 
-"@redis/search@1.1.2":
-  version "1.1.2"
-  resolved "https://registry.yarnpkg.com/@redis/search/-/search-1.1.2.tgz#6a8f66ba90812d39c2457420f859ce8fbd8f3838"
-  integrity sha512-/cMfstG/fOh/SsE+4/BQGeuH/JJloeWuH+qJzM8dbxuWvdWibWAOAHHCZTMPhV3xIlH4/cUEIA8OV5QnYpaVoA==
-
 "@redis/time-series@1.0.3":
   version "1.0.3"
   resolved "https://registry.yarnpkg.com/@redis/time-series/-/time-series-1.0.3.tgz#4cfca8e564228c0bddcdf4418cba60c20b224ac4"
   integrity sha512-OFp0q4SGrTH0Mruf6oFsHGea58u8vS/iI5+NpYdicaM+7BgqBZH8FFvNZ8rYYLrUO/QRqMq72NpXmxLVNcdmjA==
 
-"@redis/time-series@1.0.4":
-  version "1.0.4"
-  resolved "https://registry.yarnpkg.com/@redis/time-series/-/time-series-1.0.4.tgz#af85eb080f6934580e4d3b58046026b6c2b18717"
-  integrity sha512-ThUIgo2U/g7cCuZavucQTQzA9g9JbDDY2f64u3AbAoz/8vE2lt2U37LamDUVChhaDA3IRT9R6VvJwqnUfTJzng==
-
 "@remusao/guess-url-type@^1.1.2":
   version "1.2.1"
   resolved "https://registry.yarnpkg.com/@remusao/guess-url-type/-/guess-url-type-1.2.1.tgz#b3e7c32abdf98d0fb4f93cc67cad580b5fe4ba57"
@@ -11841,11 +11812,6 @@ cluster-key-slot@1.1.0:
   resolved "https://registry.yarnpkg.com/cluster-key-slot/-/cluster-key-slot-1.1.0.tgz#30474b2a981fb12172695833052bc0d01336d10d"
   integrity sha512-2Nii8p3RwAPiFwsnZvukotvow2rIHM+yQ6ZcBXGHdniadkYGZYiGmkHJIbZPIV9nfv7m/U1IPMVVcAhoWFeklw==
 
-cluster-key-slot@1.1.2:
-  version "1.1.2"
-  resolved "https://registry.yarnpkg.com/cluster-key-slot/-/cluster-key-slot-1.1.2.tgz#88ddaa46906e303b5de30d3153b7d9fe0a0c19ac"
-  integrity sha512-RMr0FhtfXemyinomL4hrWcYJxmX6deFdCxpJzhDttxgO1+bcCnkk+9drydLVDmAMG7NE6aN/fl4F7ucU/90gAA==
-
 cmd-shim@^4.1.0:
   version "4.1.0"
   resolved "https://registry.yarnpkg.com/cmd-shim/-/cmd-shim-4.1.0.tgz#b3a904a6743e9fede4148c6f3800bf2a08135bdd"
@@ -15354,11 +15320,6 @@ generic-pool@3.8.2:
   resolved "https://registry.yarnpkg.com/generic-pool/-/generic-pool-3.8.2.tgz#aab4f280adb522fdfbdc5e5b64d718d3683f04e9"
   integrity sha512-nGToKy6p3PAbYQ7p1UlWl6vSPwfwU6TMSWK7TTu+WUY4ZjyZQGniGGt2oNVvyNSpyZYSB43zMXVLcBm08MTMkg==
 
-generic-pool@3.9.0:
-  version "3.9.0"
-  resolved "https://registry.yarnpkg.com/generic-pool/-/generic-pool-3.9.0.tgz#36f4a678e963f4fdb8707eab050823abc4e8f5e4"
-  integrity sha512-hymDOu5B53XvN4QT9dBmZxPX4CWhBPPLguTZ9MMFeFa/Kg0xWVfylOVNlJji/E7yTZWFd/q9GO5TxDLq156D7g==
-
 gensync@^1.0.0-beta.1, gensync@^1.0.0-beta.2:
   version "1.0.0-beta.2"
   resolved "https://registry.yarnpkg.com/gensync/-/gensync-1.0.0-beta.2.tgz#32a6ee76c3d7f52d46b2b1ae5d93fea8580a25e0"
@@ -24140,18 +24101,6 @@ redis@^4.3.1:
     "@redis/search" "1.1.0"
     "@redis/time-series" "1.0.3"
 
-redis@^4.6.6:
-  version "4.6.6"
-  resolved "https://registry.yarnpkg.com/redis/-/redis-4.6.6.tgz#46d4f2d149d1634d6ef53db5747412a0ef7974ec"
-  integrity sha512-aLs2fuBFV/VJ28oLBqYykfnhGGkFxvx0HdCEBYdJ99FFbSEMZ7c1nVKwR6ZRv+7bb7JnC0mmCzaqu8frgOYhpA==
-  dependencies:
-    "@redis/bloom" "1.2.0"
-    "@redis/client" "1.5.7"
-    "@redis/graph" "1.1.0"
-    "@redis/json" "1.0.4"
-    "@redis/search" "1.1.2"
-    "@redis/time-series" "1.0.4"
-
 reflect-metadata@^0.1.13:
   version "0.1.13"
   resolved "https://registry.yarnpkg.com/reflect-metadata/-/reflect-metadata-0.1.13.tgz#67ae3ca57c972a2aa1642b10fe363fe32d49dc08"