Adding timer source

Author: GTFalcao

components/airtop/airtop.app.mjs

@@ -15,6 +15,7 @@ export default {
 
         const data = await this.listSessions({
           params: {
+            status: "running",
             limit,
             offset,
           },
@@ -124,15 +125,9 @@ export default {
         ...args,
       });
     },
-    async listSessions({
-      params, ...args
-    }) {
+    async listSessions(args = {}) {
       return this._makeRequest({
         url: "/sessions",
-        params: {
-          status: "running",
-          ...params,
-        },
         ...args,
       });
     },

components/airtop/sources/common/polling.mjs

@@ -0,0 +1,53 @@
+import { DEFAULT_POLLING_SOURCE_TIMER_INTERVAL } from "@pipedream/platform";
+import airtop from "../../airtop.app.mjs";
+
+export default {
+  props: {
+    airtop,
+    db: "$.service.db",
+    timer: {
+      type: "$.interface.timer",
+      default: {
+        intervalSeconds: DEFAULT_POLLING_SOURCE_TIMER_INTERVAL,
+      },
+    },
+  },
+  methods: {
+    _getLastTs() {
+      return this.db.get("lastTs");
+    },
+    _setLastTs(lastTs) {
+      this.db.set("lastTs", lastTs);
+    },
+    isNew(resource, lastTs) {
+      if (!resource.dateCreated || !lastTs) {
+        return true;
+      }
+      return new Date(resource.dateCreated).getTime() > lastTs;
+    },
+    getResources() {
+      throw new Error("getResources is not implemented");
+    },
+    generateMeta() {
+      throw new Error("generateMeta is not implemented");
+    },
+  },
+  async run() {
+    let lastTs = this._getLastTs();
+
+    const resources = await this.getResources(lastTs);
+    for (const resource of resources) {
+      const { dateCreated } = resource;
+      if (!lastTs || (dateCreated && new Date(dateCreated).getTime() > lastTs)) {
+        lastTs = new Date(dateCreated).getTime();
+      }
+      const meta = this.generateMeta(resource);
+      this.$emit(resource, meta);
+    }
+
+    if (lastTs) {
+      this._setLastTs(lastTs);
+    }
+  },
+};
+

components/airtop/sources/new-session-created/new-session-created.mjs

@@ -0,0 +1,59 @@
+import common from "../common/polling.mjs";
+
+export default {
+  ...common,
+  key: "airtop-new-session-created",
+  name: "New Session Created",
+  description: "Emit new event when a new session is created in Airtop. [See the documentation](https://docs.airtop.ai/api-reference/airtop-api/sessions/list)",
+  version: "0.0.1",
+  type: "source",
+  dedupe: "unique",
+  methods: {
+    ...common.methods,
+    async getResources(lastTs) {
+      const resources = [];
+      let hasMore = true;
+      let offset = 0;
+      const limit = 25;
+      const isFirstRun = !lastTs;
+
+      while (hasMore) {
+        const data = await this.airtop.listSessions({
+          params: {
+            limit,
+            offset,
+          },
+        });
+
+        if (!data?.sessions?.length) {
+          break;
+        }
+
+        for (const resource of data.sessions) {
+          const isNewResource = this.isNew(resource, lastTs);
+          if (isNewResource) {
+            resources.push(resource);
+          }
+        }
+
+        hasMore = data.pagination?.hasMore;
+        offset = data.pagination?.nextOffset;
+
+        // Stop on first run or on last page
+        if (isFirstRun || !hasMore) {
+          break;
+        }
+      }
+
+      return resources;
+    },
+    generateMeta(session) {
+      return {
+        id: session.id,
+        summary: `New Session: ${session.id.slice(0, 8)}`,
+        ts: new Date(session.dateCreated).getTime(),
+      };
+    },
+  },
+};
+