[BUG] Airtable triggers are getting OOM

Author: jcortes

components/airtable_oauth/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@pipedream/airtable_oauth",
-  "version": "0.4.1",
+  "version": "0.4.2",
   "description": "Pipedream Airtable (OAuth) Components",
   "main": "airtable_oauth.app.mjs",
   "keywords": [

components/airtable_oauth/sources/common/common.mjs

@@ -42,5 +42,12 @@ export default {
       const formattedTimestamp = new Date(timestampMillis).toISOString();
       this._setLastTimestamp(formattedTimestamp);
     },
+    getListRecordsParams(params) {
+      return {
+        filterByFormula: `LAST_MODIFIED_TIME() > "${this._getLastTimestamp()}"`,
+        returnFieldsByFieldId: this.returnFieldsByFieldId || false,
+        ...params,
+      };
+    },
   },
 };

components/airtable_oauth/sources/new-modified-or-deleted-records/new-modified-or-deleted-records.mjs

@@ -6,7 +6,7 @@ export default {
   name: "New, Modified or Deleted Records",
   description: "Emit new event each time a record is added, updated, or deleted in an Airtable table. Supports tables up to 10,000 records",
   key: "airtable_oauth-new-modified-or-deleted-records",
-  version: "0.0.8",
+  version: "0.0.9",
   type: "source",
   dedupe: "unique",
   props: {
@@ -53,38 +53,46 @@ export default {
     const prevAllRecordIds = this._getPrevAllRecordIds();
 
     const lastTimestamp = this._getLastTimestamp();
-    const params = {
-      filterByFormula: `LAST_MODIFIED_TIME() > "${lastTimestamp}"`,
-      returnFieldsByFieldId: this.returnFieldsByFieldId || false,
-    };
+    const params = this.getListRecordsParams();
 
     const records = await this.airtable.listRecords({
       baseId,
       tableId,
       params,
     });
 
-    let allRecordIds = [],
-      newRecordsCount = 0,
+    let newRecordsCount = 0,
       modifiedRecordsCount = 0,
       deletedRecordsCount = 0;
 
     if (records) {
       for (const record of records) {
-        if (!lastTimestamp || moment(record.createdTime) > moment(lastTimestamp)) {
-          record.type = "new_record";
+        if (!lastTimestamp || moment(record.createdTime) > moment(lastTimestamp)) {;
+          this.$emit({
+            ...record,
+            type: "new_record",
+            metadata,
+          }, {
+            id: record.id,
+            summary: `New record: ${record.id}`,
+            ts: moment(record.createdTime).valueOf(),
+          });
           newRecordsCount++;
+
         } else {
-          record.type = "record_modified";
+          const ts = Date.now();
+          const id = `${record.id}-${ts}`;
+          this.$emit({
+            ...record,
+            type: "record_modified",
+            metadata,
+          }, {
+            id,
+            summary: `Record modified: ${record.id}`,
+            ts,
+          });
           modifiedRecordsCount++;
         }
-
-        record.metadata = metadata;
-
-        this.$emit(record, {
-          summary: `${record.type}: ${JSON.stringify(record.fields)}`,
-          id: record.id,
-        });
       }
     }
 
@@ -95,26 +103,27 @@ export default {
       tableId,
       params,
     });
-    if (!data.length || data.length === 0) return;
-    allRecordIds = [
-      ...data.map((record) => record.id),
-    ];
+
+    const allRecordIds = data.map((record) => record.id);
 
     if (prevAllRecordIds) {
-      const deletedRecordIds = prevAllRecordIds.filter(
-        (prevRecord) => !allRecordIds.includes(prevRecord),
-      );
+      const currentRecordIdSet = new Set(allRecordIds);
+      const deletedRecordIds =
+        prevAllRecordIds.filter((prevRecord) => !currentRecordIdSet.has(prevRecord));
+
       for (const recordID of deletedRecordIds) {
-        deletedRecordsCount++;
-        const deletedRecordObj = {
+        const ts = Date.now();
+        const id = `${recordID}-${ts}`;
+        this.$emit({
+          id: recordID,
           metadata,
           type: "record_deleted",
-          id: recordID,
-        };
-        this.$emit(deletedRecordObj, {
+        }, {
+          id,
           summary: `Record deleted: ${recordID}`,
-          id: recordID,
+          ts,
         });
+        deletedRecordsCount++;
       }
     }
 

components/airtable_oauth/sources/new-or-modified-records-in-view/new-or-modified-records-in-view.mjs

@@ -6,7 +6,7 @@ export default {
   name: "New or Modified Records in View",
   description: "Emit new event for each new or modified record in a view",
   key: "airtable_oauth-new-or-modified-records-in-view",
-  version: "0.0.9",
+  version: "0.0.10",
   type: "source",
   props: {
     ...base.props,
@@ -48,11 +48,9 @@ export default {
     } = this;
 
     const lastTimestamp = this._getLastTimestamp();
-    const params = {
+    const params = this.getListRecordsParams({
       view: viewId,
-      filterByFormula: `LAST_MODIFIED_TIME() > "${lastTimestamp}"`,
-      returnFieldsByFieldId: this.returnFieldsByFieldId || false,
-    };
+    });
 
     const records = await this.airtable.listRecords({
       baseId,
@@ -74,19 +72,31 @@ export default {
     let newRecords = 0, modifiedRecords = 0;
     for (const record of records) {
       if (!lastTimestamp || moment(record.createdTime) > moment(lastTimestamp)) {
-        record.type = "new_record";
+        this.$emit({
+          ...record,
+          type: "new_record",
+          metadata,
+        }, {
+          id: record.id,
+          summary: `New record: ${record.id}`,
+          ts: moment(record.createdTime).valueOf(),
+        });
         newRecords++;
+
       } else {
-        record.type = "record_modified";
+        const ts = Date.now();
+        const id = `${record.id}-${ts}`;
+        this.$emit({
+          ...record,
+          type: "record_modified",
+          metadata,
+        }, {
+          id,
+          summary: `Record modified: ${record.id}`,
+          ts,
+        });
         modifiedRecords++;
       }
-
-      record.metadata = metadata;
-
-      this.$emit(record, {
-        summary: `${record.type}: ${JSON.stringify(record.fields)}`,
-        id: record.id,
-      });
     }
     console.log(`Emitted ${newRecords} new records(s) and ${modifiedRecords} modified record(s).`);
 

components/airtable_oauth/sources/new-records-in-view/new-records-in-view.mjs

@@ -6,7 +6,7 @@ export default {
   name: "New Records in View",
   description: "Emit new event for each new record in a view",
   key: "airtable_oauth-new-records-in-view",
-  version: "0.0.8",
+  version: "0.0.9",
   type: "source",
   dedupe: "unique",
   props: {
@@ -49,11 +49,10 @@ export default {
     } = this;
 
     const lastTimestamp = this._getLastTimestamp();
-    const params = {
+    const params = this.getListRecordsParams({
       view: viewId,
       filterByFormula: `CREATED_TIME() > "${lastTimestamp}"`,
-      returnFieldsByFieldId: this.returnFieldsByFieldId || false,
-    };
+    });
 
     const records = await this.airtable.listRecords({
       baseId,