feat: Fetch data while considering previously persisted records

packages/backend/src/apps/twitter/triggers/search-tweet.ts

@@ -16,12 +16,43 @@ export default class SearchTweet {
     this.parameters = parameters;
   }
 
-  async run() {
-    const response = await this.client.v2.get('tweets/search/recent', {
-      query: this.parameters.searchTerm as string,
-      max_results: 10,
-    });
+  async run(startTime: Date) {
+    const tweets = [];
 
-    return response.data;
+    const response = await this.client.v2.search(
+      this.parameters.searchTerm as string,
+      {
+        max_results: 50,
+        'tweet.fields': 'created_at',
+      }
+    );
+
+    for await (const tweet of response.data.data) {
+      if (new Date(tweet.created_at).getTime() <= startTime.getTime()) {
+        break;
+      }
+
+      tweets.push(tweet);
+
+      if (response.data.meta.next_token) {
+        await response.fetchNext();
+      }
+    }
+
+    return tweets;
+  }
+
+  async testRun() {
+    const response = await this.client.v2.search(
+      this.parameters.searchTerm as string,
+      {
+        max_results: 10,
+        'tweet.fields': 'created_at',
+      }
+    );
+
+    const mostRecentTweet = response.data.data[0];
+
+    return [mostRecentTweet];
   }
 }

packages/backend/src/models/flow.ts

@@ -1,5 +1,5 @@
 import { ValidationError } from 'objection';
-import type { QueryContext, ModelOptions } from 'objection';
+import type { ModelOptions } from 'objection';
 import Base from './base';
 import Step from './step';
 import Execution from './execution';

packages/backend/src/services/processor.ts

@@ -121,8 +121,26 @@ class Processor {
       rawParameters
     );
 
+    const lastExecutionStep = await step
+      .$relatedQuery('executionSteps')
+      .orderBy('created_at', 'desc')
+      .first();
+
+    const lastExecutionStepCratedAt = lastExecutionStep?.dataOut
+      ?.created_at as string;
+    const flow = (await step.$relatedQuery('flow')) as Flow;
+
     const command = appInstance.triggers[key];
-    const fetchedData = await command.run();
+
+    const startTime = new Date(lastExecutionStepCratedAt || flow.updatedAt);
+    let fetchedData;
+
+    if (this.testRun) {
+      fetchedData = await command.testRun(startTime);
+    } else {
+      fetchedData = await command.run(startTime);
+    }
+
     return fetchedData;
   }