add new message source

Author: SokolovskyiK

components/drift/common/utils.mjs

@@ -22,6 +22,27 @@ const removeNullEntries = (obj) =>
       : acc;
   }, {});
 
+function doesContextMatch(inputContext, fetchedContext) {
+
+  if (typeof inputContext !== "object" || inputContext === null || Array.isArray(inputContext)) {
+    throw new Error ("Message context is not an object");
+  };
+
+  for (const key of Object.keys(inputContext)) {
+    if (!(key in fetchedContext)) {
+      console.log(`Invalid context field "${key}", emission skipped` );
+      return false;
+    }
+    if (fetchedContext[key] !== inputContext[key]) {
+      console.log(`Context values of "${key}" do not match, emission skipped` );
+      return false;
+    }
+  }
+  return true;
+};
+
 export {
   removeNullEntries,
+  doesContextMatch,
 };
+

components/drift/drift.app.mjs

@@ -126,6 +126,27 @@ export default {
       const newest = arr.slice(0, firtsNew);
       return newest.reverse();
     },
+    async getMessagesByConvId($, conversationId) {
+
+      const messages = [];
+      let next;
+
+      do {
+        const result = await this._makeRequest({
+          $,
+          path: `/conversations/${conversationId}/messages${next
+            ? `?next=${next}`
+            : ""}`,
+        });
+
+        messages.push(...result.data.messages);
+        next = result?.pagination?.next;
+
+      } while (next);
+
+      return messages;
+    },
+
     parseIfJSONString(input) {
 
       if (typeof input === "string") {

components/drift/sources/new-conversation/new-conversation.mjs

@@ -20,30 +20,6 @@ export default {
       },
     },
   },
-  hooks: {
-    async activate() {
-
-      const {
-        drift,
-        db,
-      } = this;
-
-      await db.set("lastConversation", null); //reset
-
-      const result = await drift._makeRequest({
-        path: "/conversations/list?limit=100&statusId=1",
-      });
-
-      if (!result.data.length) {
-        console.log("No conversations found.");
-        return;
-      };
-
-      await db.set("lastConversation", result.data[0].id);
-      console.log(`Initialized with ID ${result.data[0].id}.`);
-
-    },
-  },
 
   async run({ $ }) {
     const {

components/drift/sources/new-message/new-message.mjs

@@ -1,64 +1,90 @@
 import drift from "../../drift.app.mjs";
+import { DEFAULT_POLLING_SOURCE_TIMER_INTERVAL } from "@pipedream/platform";
+import { doesContextMatch } from "../../common/utils.mjs";
 
 export default {
-  key: "drift-new-message-instant",
-  name: "New Message",
-  description: "Emit new event when a new message is received in Drift.  [See the docs](https://devdocs.drift.com/docs/webhook-events-1).",
-  version: "0.0.1",
+  key: "drift-new-message",
+  name: "New Message in Conversation",
+  description: "Emit new event when a message is received in a specific Drift conversation.",
+  version: "0.0.3",
   type: "source",
+  dedupe: "unique",
+
   props: {
     drift,
-    http: "$.interface.http",
+    db: "$.service.db",
     conversationId: {
       type: "integer",
       label: "Conversation ID",
-      description: "The ID of the conversation to monitor. Emits events for all new messages if not provided.",
-      optional: true,
+      description: "Enter the ID of the conversation",
     },
-    emailOrId: {
-      type: "string",
-      label: "Email or ID",
-      description: "Email or ID of the user to monitor. Emits events for all new messages if not provided.",
+    messageContext: {
+      type: "object",
+      label: "Message Context",
+      description: "Enter message context [See the documentation](https://devdocs.drift.com/docs/message-model).",
       optional: true,
     },
+    timer: {
+      type: "$.interface.timer",
+      label: "Polling Interval",
+      description: "How often to poll Drift for new messages.",
+      default: {
+        intervalSeconds: DEFAULT_POLLING_SOURCE_TIMER_INTERVAL,
+      },
+    },
   },
-  async run(event) {
 
-    const { body } = event;
-    const { drift } = this;
+  async run({ $ }) {
+    const {
+      drift,
+      db,
+      conversationId,
+    } = this;
 
-    if (body?.type !== "new_message") {
-      console.log("Ignored non-new_message event:", body?.type);
+    const messageContext = drift.parseIfJSONString(this.messageContext);
+
+    console.log(messageContext);
+    const messages = await drift.getMessagesByConvId($, conversationId);
+
+    if (!messages?.length) {
+      console.log("No messages found.");
       return;
     };
 
-    // If conversationId provided
-    if (this.conversationId && !(this.conversationId === Number(body.data.conversationId))) {
-      console.log(`Ignored. Wrong conversationId. 
-                Expected ${this.conversationId} got ${body.data.conversationId}`);
+    let lastMessageId = await db.get("lastMessage");
+
+    const lastFetchedMsgId = messages[messages.length - 1].id;
+
+    if (!lastMessageId) {
+      await db.set("lastMessage", lastFetchedMsgId);
+      console.log(`Initialized with ID ${lastFetchedMsgId}.`);
       return;
     };
 
-    const contactId = body.data.author.id;
-
-    const result = await drift.getContactById({
-      contactId,
-    });
+    if (lastMessageId === lastFetchedMsgId) {
+      console.log("No new messages found");
+      return;
+    };
 
-    const email = result.data?.attributes?.email || "unknown";
+    const lastMessageIndex = messages.findIndex((obj) => obj.id === lastMessageId);
 
-    if (this.emailOrId &&
-            !(email === this.emailOrId || Number(contactId) === Number(this.emailOrId))) {
-      console.log(`Ignored. Wrong emailOrId. Expected ${this.emailOrId}`);
+    if (lastMessageIndex === -1) {
+      console.log("Last message ID not found.");
       return;
     };
 
-    body.data.attributes =  result.data.attributes;
+    for (let i = lastMessageIndex + 1; i < messages.length; i++) {
+      if (messageContext) {
+        if (!doesContextMatch(messageContext, messages[i].context)) continue;
+      }
+      this.$emit(messages[i], {
+        id: messages[i].id,
+        summary: `New message with ID ${messages[i].id}`,
+        ts: messages[i].createdAt,
+      });
+    };
+
+    await db.set("lastMessage", lastFetchedMsgId);
 
-    this.$emit(body, {
-      summary: `New message from contact "${email} " ID "${contactId || "unknown"}"`,
-      id: body.data.endUserId,
-      ts: body.timeStamp,
-    });
   },
 };