[BUG] Gmail (Developer App) - New Attachment source not working consistently

components/gmail_custom_oauth/actions/add-label-to-email/add-label-to-email.mjs

@@ -4,7 +4,7 @@ export default {
   key: "gmail_custom_oauth-add-label-to-email",
   name: "Add Label to Email",
   description: "Add a label to an email message. [See the docs](https://developers.google.com/gmail/api/reference/rest/v1/users.messages/modify)",
-  version: "0.0.8",
+  version: "0.0.9",
   type: "action",
   props: {
     gmail,

components/gmail_custom_oauth/actions/create-draft/create-draft.mjs

@@ -5,7 +5,7 @@ export default {
   key: "gmail_custom_oauth-create-draft",
   name: "Create Draft",
   description: "Create a draft from your Google Workspace email account",
-  version: "0.0.4",
+  version: "0.0.5",
   type: "action",
   props: {
     gmail,

components/gmail_custom_oauth/actions/download-attachment/download-attachment.mjs

@@ -6,7 +6,7 @@ export default {
   key: "gmail_custom_oauth-download-attachment",
   name: "Download Attachement",
   description: "Download an attachment by attachmentId to the /tmp directory. [See the docs](https://developers.google.com/gmail/api/reference/rest/v1/users.messages.attachments/get)",
-  version: "0.0.2",
+  version: "0.0.3",
   type: "action",
   props: {
     gmail,

components/gmail_custom_oauth/actions/find-email/find-email.mjs

@@ -4,7 +4,7 @@ export default {
   key: "gmail_custom_oauth-find-email",
   name: "Find Email",
   description: "Find an email using Google's Search Engine. [See the docs](https://developers.google.com/gmail/api/reference/rest/v1/users.messages/list)",
-  version: "0.0.9",
+  version: "0.0.10",
   type: "action",
   props: {
     gmail,

components/gmail_custom_oauth/actions/send-email/send-email.mjs

@@ -10,7 +10,7 @@ overrideApp(base);
 export default {
   ...base,
   key: "gmail_custom_oauth-send-email",
-  version: "0.1.1",
+  version: "0.1.2",
   props: {
     ...base.props,
     inReplyTo: {

components/gmail_custom_oauth/actions/update-org-signature/update-org-signature.mjs

@@ -7,7 +7,7 @@ export default {
   name: "Update Signature for Email in Organization",
   description: `Update the signature for a specific email address in an organization.
     A Google Cloud service account with delegated domain-wide authority is required for this action. [See docs here](https://developers.google.com/gmail/api/reference/rest/v1/users.settings.sendAs/update)`,
-  version: "0.0.5",
+  version: "0.0.6",
   type: "action",
   props: {
     gmail: base.props.gmail,

components/gmail_custom_oauth/actions/update-primary-signature/update-primary-signature.mjs

@@ -4,7 +4,7 @@ export default {
   key: "gmail_custom_oauth-update-primary-signature",
   name: "Update Signature for Primary Email Address",
   description: "Update the signature for the primary email address. [See docs here](https://developers.google.com/gmail/api/reference/rest/v1/users.settings.sendAs/update)",
-  version: "0.0.5",
+  version: "0.0.6",
   type: "action",
   props: {
     gmail,

components/gmail_custom_oauth/gmail_custom_oauth.app.mjs

@@ -91,5 +91,24 @@ export default {
       });
       return data;
     },
+    async getMessagesWithRetry(ids = [], retryCount = 0) {
+      const promises = ids.map((id) =>
+        this.getMessage({
+          id,
+        })
+          .catch((err) => {
+            console.error(`Failed to get message with id ${id}:`, err);
+            if (retryCount < 3) {
+              console.log("Retrying...");
+              return this.getMessagesWithRetry([
+                id,
+              ], retryCount + 1);
+            }
+
+            console.error("Failed after 3 attempts.");
+            return null;
+          }));
+      return Promise.all(promises);
+    },
   },
 };

components/gmail_custom_oauth/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@pipedream/gmail_custom_oauth",
-  "version": "0.1.7",
+  "version": "0.1.8",
   "description": "Pipedream Gmail (Consumer) Components",
   "main": "gmail_custom_oauth.app.mjs",
   "keywords": [

components/gmail_custom_oauth/sources/common/polling.mjs

@@ -28,53 +28,45 @@ export default {
       optional: true,
     },
   },
-  hooks: {
-    async deploy() {
-      console.log(`Fetching last ${constants.HISTORICAL_EVENTS} historical events...`);
-      const response = await this.gmail.listMessages({
-        q: this.q,
-        labelIds: this.labels,
-        maxResults: constants.HISTORICAL_EVENTS,
-      });
-
-      const messageIds = response?.messages?.map((message) => message.id);
-
-      if (messageIds?.length) {
-        this.setLastMessageId(messageIds[0]);
-        await this.processMessageIds(messageIds.reverse());
-      }
-    },
-  },
   methods: {
     getLastMessageId() {
       return this.db.get("lastMessageId");
     },
     setLastMessageId(lastMessageId) {
-      if (lastMessageId) {
-        this.db.set("lastMessageId", lastMessageId);
-      }
+      this.db.set("lastMessageId", lastMessageId);
     },
     processMessageIds() {
       throw new Error("processMessageIds not implemented");
     },
+    constructQuery() {
+      return this.q;
+    },
   },
   async run() {
-    const lastMessageId = this.getLastMessageId();
+    const {
+      gmail,
+      constructQuery,
+      labels,
+    } = this;
 
-    const response = await this.gmail.listMessages({
-      q: this.q,
-      labelIds: this.labels,
-      maxResults: 100,
+    console.log("Polling for new messages...");
+    const { messages } = await gmail.listMessages({
+      q: constructQuery(),
+      labelIds: labels,
+      maxResults: constants.HISTORICAL_EVENTS,
     });
 
-    let messageIds = response?.messages?.map((message) => message.id);
+    let messageIds = messages?.map((message) => message.id);
 
     if (!messageIds?.length) {
-      console.log("No new message. Exiting...");
+      console.log("There are no new messages. Exiting...");
       return;
     }
 
-    this.setLastMessageId(messageIds[0]);
+    const lastMessageId = this.getLastMessageId();
+    if (lastMessageId !== messageIds[0]) {
+      this.setLastMessageId(messageIds[0]);
+    }
 
     const index = messageIds.indexOf(lastMessageId);
     if (index !== -1) {
@@ -83,7 +75,7 @@ export default {
 
     const numMessages = messageIds.length;
     if (!numMessages) {
-      console.log("No new message. Exiting...");
+      console.log("Messages already processed. Exiting...");
       return;
     }
 

components/gmail_custom_oauth/sources/new-attachment-received/new-attachment-received.mjs

@@ -5,45 +5,44 @@ export default {
   key: "gmail_custom_oauth-new-attachment-received",
   name: "New Attachment Received",
   description: "Emit new event for each attachment in a message received. This source is capped at 100 max new messages per run.",
-  version: "0.0.7",
+  version: "0.0.8",
   type: "source",
   dedupe: "unique",
   methods: {
     ...common.methods,
+    constructQuery() {
+      const { q: query } = this;
+      const hasAttachment = "has:attachment";
+      return query?.includes(hasAttachment)
+        ? query
+        : [
+          hasAttachment,
+          query,
+        ].join(" ").trim();
+    },
     generateMeta(attachment, message) {
       return {
         id: attachment.body.attachmentId,
-        summary: `New attachment: ${attachment.filename}`,
+        summary: `New Attachment: ${attachment.filename}`,
         ts: message.internalDate,
       };
     },
     emitEvents(messages) {
-      for (const message of messages) {
-        const attachments = message.payload.parts.filter((part) => part.body.attachmentId);
-        const numAttachments = attachments.length;
-        if (!numAttachments) continue;
-        const suffix = numAttachments === 1
-          ? ""
-          : "s";
-        console.log(`Emitting event${suffix} for ${numAttachments} attachment${suffix} found for message`);
-        for (const attachment of attachments) {
-          const meta = this.generateMeta(attachment, message);
-          this.$emit({
-            message,
-            attachment,
-          }, meta);
+      messages?.forEach((message) => {
+        if (message) {
+          const { parts: attachments } = message.payload;
+
+          attachments.forEach((attachment) => {
+            this.$emit({
+              message,
+              attachment,
+            }, this.generateMeta(attachment, message));
+          });
         }
-      }
-    },
-    filterMessagesWithAttachments(messages) {
-      return messages.filter(
-        (message) => message.payload.parts?.filter((part) => part.body?.attachmentId).length,
-      );
+      });
     },
     async processMessageIds(messageIds) {
-      const messages = this.filterMessagesWithAttachments(
-        await this.gmail.getMessages(messageIds),
-      );
+      const messages = await this.gmail.getMessagesWithRetry(messageIds);
       this.emitEvents(messages);
     },
   },

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

@@ -5,7 +5,7 @@ export default {
   key: "gmail_custom_oauth-new-email-received",
   name: "New Email Received",
   description: "Emit new event when an email is received. This source is capped at 100 max new messages per run.",
-  version: "0.0.8",
+  version: "0.0.9",
   type: "source",
   dedupe: "unique",
   methods: {

components/gmail_custom_oauth/sources/new-labeled-email/new-labeled-email.mjs

@@ -7,7 +7,7 @@ export default {
   name: "New Labeled Email",
   description: "Emit new event when a new email is labeled.",
   type: "source",
-  version: "0.0.5",
+  version: "0.0.6",
   dedupe: "unique",
   props: {
     gmail,

components/gmail_custom_oauth/sources/new-sent-email/new-sent-email.mjs

@@ -5,7 +5,7 @@ export default {
   key: "gmail_custom_oauth-new-sent-email",
   name: "New Sent Email",
   description: "Emit new event for each new email sent. (Maximum of 300 events emited per execution)",
-  version: "0.0.4",
+  version: "0.0.5",
   type: "source",
   props: {
     app,