feat: upgrade to response-rejection plugin

.changeset/empty-crews-attack.md

@@ -0,0 +1,6 @@
+---
+'@flatfile/util-response-rejection': patch
+'@flatfile/plugin-webhook-egress': patch
+---
+
+Upgrade to the @flatfile/util-response-rejection plugin to support deleting successfully submitted records or adding a status column to indicate successful/rejected records.

plugins/webhook-egress/src/webhook.egress.ts

@@ -2,11 +2,11 @@ import api from '@flatfile/api'
 import { FlatfileListener } from '@flatfile/listener'
 import { jobHandler } from '@flatfile/plugin-job-handler'
 import { logError } from '@flatfile/util-common'
+import axios from 'axios'
 import {
-  ResponseRejection,
+  RejectionResponse,
   responseRejectionHandler,
-} from '@flatfile/util-response-rejection'
-import axios from 'axios'
+} from '../../../utils/response-rejection/src' // TODO: replace with '@flatfile/util-response-rejection'
 
 export function webhookEgress(job: string, webhookUrl?: string) {
   return function (listener: FlatfileListener) {
@@ -48,22 +48,11 @@ export function webhookEgress(job: string, webhookUrl?: string) {
           )
 
           if (response.status === 200) {
-            const rejections: ResponseRejection = response.data.rejections
+            const rejections: RejectionResponse = response.data.rejections
             if (rejections) {
-              const totalRejectedRecords = await responseRejectionHandler(
-                rejections
-              )
-              return {
-                outcome: {
-                  next: {
-                    type: 'id',
-                    id: rejections.id,
-                    label: 'See rejections',
-                  },
-                  message: `${totalRejectedRecords} record(s) were rejected during data submission. Review the rejection notes, fix, then resubmit.`,
-                },
-              }
+              return await responseRejectionHandler(rejections)
             }
+
             return {
               outcome: {
                 message: `Data was successfully submitted to the provided webhook. Go check it out at ${webhookReceiver}.`,
@@ -85,13 +74,8 @@ export function webhookEgress(job: string, webhookUrl?: string) {
             '@flatfile/plugin-webhook-egress',
             JSON.stringify(error, null, 2)
           )
-
-          return {
-            outcome: {
-              message:
-                "This job failed probably because it couldn't find the webhook URL.",
-            },
-          }
+          // Throw error to fail job
+          throw new Error(`Error posting data to webhook`)
         }
       })
     )

utils/response-rejection/package.json

@@ -27,6 +27,7 @@
   },
   "license": "ISC",
   "dependencies": {
-    "@flatfile/api": "^1.5.37"
+    "@flatfile/api": "^1.5.37",
+    "@flatfile/util-common": "^0.2.3"
   }
 }

utils/response-rejection/src/index.ts

@@ -1,63 +1,156 @@
 import api, { Flatfile } from '@flatfile/api'
+import { processRecords } from '@flatfile/util-common'
 
-export interface ResponseRejection {
+export interface RejectionResponse {
   id: string
-  sheets: ResponseRejectionSheet[]
+  message?: string
+  deleteSubmitted?: boolean
+  sheets: SheetRejections[]
 }
 
-export interface ResponseRejectionSheet {
-  id: string
-  name?: string
-  rejectedRecords: ResponseRejectionRecord[]
+export interface SheetRejections {
+  sheetId: string
+  rejectedRecords: RecordRejections[]
 }
 
-export interface ResponseRejectionRecord {
+export interface RecordRejections {
   id: string
   values: { field: string; message: string }[]
 }
 
 export async function responseRejectionHandler(
-  responseRejection: ResponseRejection
-): Promise<number> {
+  responseRejection: RejectionResponse
+): Promise<void | Flatfile.JobCompleteDetails> {
   let totalRejectedRecords = 0
 
-  const results = await Promise.all(
-    responseRejection.sheets.map((sheet) => updateSheet(sheet))
-  )
-  totalRejectedRecords = results.reduce((acc, val) => acc + val, 0)
+  // Using a for...of loop to handle asynchronous operations
+  for (const sheet of responseRejection.sheets || []) {
+    const count = await updateSheet(sheet, responseRejection.deleteSubmitted)
+    totalRejectedRecords += count
+  }
+
+  const message = responseRejection.message ?? getMessage(totalRejectedRecords)
+  const next = getNext(totalRejectedRecords)
+
+  return {
+    outcome: {
+      heading: totalRejectedRecords > 0 ? 'Rejected Records' : 'Success!',
+      acknowledge: true,
+      ...(next && { next }),
+      message,
+    },
+  }
+}
+
+function getMessage(totalRejectedRecords) {
+  return totalRejectedRecords > 0
+    ? `${totalRejectedRecords} record(s) were rejected during data submission. Review the rejection notes, fix, then resubmit.`
+    : 'Data was successfully submitted.'
+}
 
-  return totalRejectedRecords
+function getNext(totalRejectedRecords): Flatfile.JobOutcomeNext | undefined {
+  return totalRejectedRecords > 0
+    ? {
+        type: 'url',
+        url: '?filter=error',
+        label: 'See rejections',
+      }
+    : undefined
 }
 
-async function updateSheet(sheet: ResponseRejectionSheet): Promise<number> {
-  if (!sheet.rejectedRecords?.length) {
-    return 0
+async function updateSheet(
+  sheetRejections: SheetRejections,
+  deleteSubmitted: boolean
+): Promise<number> {
+  if (!deleteSubmitted) {
+    await addSubmissionStatusField(sheetRejections.sheetId)
   }
 
-  const rejectedRecordsIds = sheet.rejectedRecords.map((record) => record.id)
-  const sheetRecords = await api.records.get(sheet.id)
+  await processRecords(sheetRejections.sheetId, async (records) => {
+    await updateRecords(sheetRejections, records, deleteSubmitted)
+  })
 
-  const rejectedSheetRecords: Flatfile.Record_[] =
-    sheetRecords.data.records?.filter((record: Flatfile.Record_) =>
-      rejectedRecordsIds.includes(record.id)
-    )
+  return sheetRejections.rejectedRecords.length
+}
+
+async function addSubmissionStatusField(sheetId: string): Promise<void> {
+  try {
+    const { data: sheet } = await api.sheets.get(sheetId)
+    if (
+      !sheet.config.fields.some((field) => field.key === 'submissionStatus')
+    ) {
+      await api.sheets.addField(sheet.id, {
+        key: 'submissionStatus',
+        label: 'Submission Status',
+        type: 'enum',
+        readonly: true,
+        config: {
+          allowCustom: false,
+          options: [
+            { label: 'Rejected', value: 'rejected' },
+            { label: 'Submitted', value: 'submitted' },
+          ],
+        },
+      })
+    }
+  } catch (error) {
+    console.error('Error adding rejection status field:', error)
+    throw 'Error adding rejection status field'
+  }
+}
 
-  for (const record of rejectedSheetRecords || []) {
-    const rejectedRecord: ResponseRejectionRecord = sheet.rejectedRecords.find(
+async function updateRecords(
+  rejections: SheetRejections,
+  records: Flatfile.RecordsWithLinks,
+  deleteSubmitted: boolean
+): Promise<void> {
+  records.forEach((record) => {
+    const rejectedRecord = rejections.rejectedRecords.find(
       (item) => item.id === record.id
     )
-    for (const value of rejectedRecord.values) {
+
+    rejectedRecord?.values.forEach((value) => {
       if (record.values[value.field]) {
         record.values[value.field].messages = [
-          {
-            type: 'error',
-            message: value.message,
-          },
+          { type: 'error', message: value.message },
         ]
       }
+    })
+
+    if (!deleteSubmitted) {
+      record.values['submissionStatus'].value = rejectedRecord
+        ? 'rejected'
+        : 'submitted'
     }
+  })
+
+  try {
+    await api.records.update(rejections.sheetId, records)
+  } catch (error) {
+    console.error('Error updating records:', error)
+    throw new Error('Error updating records')
+  }
+
+  if (deleteSubmitted && records.length !== rejections.rejectedRecords.length) {
+    await deleteValidRecords(rejections.sheetId)
   }
+}
 
-  await api.records.update(sheet.id, rejectedSheetRecords)
-  return rejectedRecordsIds.length
+async function deleteValidRecords(sheetId: string): Promise<void> {
+  try {
+    const { data: sheet } = await api.sheets.get(sheetId)
+    await api.jobs.create({
+      type: 'workbook',
+      operation: 'delete-records',
+      trigger: 'immediate',
+      source: sheet.workbookId,
+      config: {
+        sheet: sheetId,
+        filter: 'valid',
+      },
+    })
+  } catch (error) {
+    console.error('Error deleting all records:', error)
+    throw new Error('Error deleting all records')
+  }
 }