feat: update matter importer metrics

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
       )
     }
   }

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) {

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) {

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