Gmail - updates to new-email-received trigger

components/gmail/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@pipedream/gmail",
-  "version": "0.1.9",
+  "version": "0.1.10",
   "description": "Pipedream Gmail Components",
   "main": "gmail.app.mjs",
   "keywords": [

components/gmail/sources/new-email-received/new-email-received.mjs

@@ -15,7 +15,7 @@ export default {
   name: "New Email Received",
   description: "Emit new event when a new email is received.",
   type: "source",
-  version: "0.1.6",
+  version: "0.1.7",
   dedupe: "unique",
   props: {
     gmail,
@@ -121,11 +121,14 @@ export default {
         );
       }
 
-      newProps.http = "$.interface.http";
+      newProps.http = {
+        type: "$.interface.http",
+        customResponse: true,
+      };
       newProps.timer = {
         type: "$.interface.timer",
         default: {
-          intervalSeconds: 24 * 60 * 60,
+          intervalSeconds: 60 * 60,
         },
         hidden: true,
       };
@@ -203,12 +206,19 @@ export default {
 
         props.latencyWarningAlert.hidden = false;
 
-        const historyId = await this.setupGmailNotifications(topicName);
+        const {
+          historyId, expiration,
+        } = await this.setupGmailNotifications(topicName);
         newProps.initialHistoryId = {
           type: "string",
           default: historyId,
           hidden: true,
         };
+        newProps.expiration = {
+          type: "string",
+          default: expiration,
+          hidden: true,
+        };
       }
     }
     props.label.hidden = false;
@@ -279,6 +289,18 @@ export default {
     _setLastProcessedHistoryId(lastProcessedHistoryId) {
       this.db.set("lastProcessedHistoryId", lastProcessedHistoryId);
     },
+    _getExpiration() {
+      return this.db.get("expiration");
+    },
+    _setExpiration(expiration) {
+      this.db.set("expiration", expiration);
+    },
+    _getLastReceivedTime() {
+      return this.db.get("lastReceivedTime");
+    },
+    _setLastReceivedTime(lastReceivedTime) {
+      this.db.set("lastReceivedTime", lastReceivedTime);
+    },
     sdkParams() {
       const authKeyJSON = JSON.parse(this.serviceAccountKeyJson);
       const {
@@ -338,7 +360,7 @@ export default {
         },
       });
       console.log("Watch response:", watchResponse);
-      return watchResponse.historyId;
+      return watchResponse;
     },
     async getOrCreateTopic(name) {
       const sdkParams = this.sdkParams();
@@ -413,17 +435,32 @@ export default {
         // event was triggered by timer
         const topicName = this._getTopicName();
         if (topicName) {
-          // renew Gmail push notifications
-          await this.setupGmailNotifications(topicName);
+          // renew Gmail push notifications if expiring within the next hour
+          // or if no email has been received within the last hour
+          const currentExpiration = this._getExpiration();
+          const lastReceivedTime = this._getLastReceivedTime();
+          if (
+            (+currentExpiration < (event.timestamp + 3600) * 1000)
+            || (lastReceivedTime < (event.timestamp - 3600) * 1000)
+          ) {
+            const { expiration } = await this.setupGmailNotifications(topicName);
+            this._setExpiration(expiration);
+          }
           return;
         } else {
           // first run, no need to renew push notifications
           this._setTopicName(this.topic);
-          this._setLastProcessedHistoryId(this.initialHistoryId);
+          const initialHistoryId = this.initialHistoryId || this._getLastHistoryId();
+          this._setLastProcessedHistoryId(initialHistoryId);
+          this._setExpiration(this.expiration);
           return;
         }
       }
 
+      this.http.respond({
+        status: 200,
+      });
+
       // Extract the Pub/Sub message data
       const pubsubMessage = event.body.message;
       if (!pubsubMessage) {
@@ -491,6 +528,8 @@ export default {
       this._setLastProcessedHistoryId(latestHistoryId);
       console.log("Updated lastProcessedHistoryId:", latestHistoryId);
 
+      this._setLastReceivedTime(Date.now());
+
       messageDetails.forEach((message) => {
         if (message?.id) {
           this.emitEvent(message);