feat: airtable integration

config/config.json

@@ -0,0 +1,11 @@
+{
+    "development": {
+      "username": "postgres",
+      "password": "postgres",
+      "database": "maxun",
+      "host": "localhost",
+      "port": 5432,
+      "dialect": "postgres"
+    }
+  }
+

migrations/20250111140925-add-airtable-fields-to-robot.js

@@ -0,0 +1,35 @@
+'use strict';
+
+/** @type {import('sequelize-cli').Migration} */
+module.exports = {
+  async up(queryInterface, Sequelize) {
+    // Add new Airtable-related columns to the 'robot' table
+    await queryInterface.addColumn('robot', 'airtable_base_id', {
+      type: Sequelize.STRING,
+      allowNull: true,
+    });
+
+    await queryInterface.addColumn('robot', 'airtable_table_name', {
+      type: Sequelize.STRING,
+      allowNull: true,
+    });
+
+    await queryInterface.addColumn('robot', 'airtable_api_key', {
+      type: Sequelize.STRING,
+      allowNull: true,
+    });
+
+    await queryInterface.addColumn('robot', 'airtable_access_token', {
+      type: Sequelize.STRING,
+      allowNull: true,
+    });
+  },
+
+  async down(queryInterface, Sequelize) {
+    // Remove Airtable-related columns from the 'robot' table
+    await queryInterface.removeColumn('robot', 'airtable_base_id');
+    await queryInterface.removeColumn('robot', 'airtable_table_name');
+    await queryInterface.removeColumn('robot', 'airtable_api_key');
+    await queryInterface.removeColumn('robot', 'airtable_access_token');
+  },
+};

package.json

@@ -21,6 +21,7 @@
     "@types/react": "^18.0.5",
     "@types/react-dom": "^18.0.1",
     "@types/uuid": "^8.3.4",
+    "airtable": "^0.12.2",
     "axios": "^0.26.0",
     "bcrypt": "^5.1.1",
     "body-parser": "^1.20.3",

server/src/models/Robot.ts

@@ -25,6 +25,10 @@ interface RobotAttributes {
   google_sheet_id?: string | null;
   google_access_token?: string | null;
   google_refresh_token?: string | null;
+  airtable_base_id?: string | null; // Airtable Base ID
+  airtable_table_name?: string | null; // Airtable Table Name
+  airtable_personal_access_token?: string | null; // Airtable Personal Access Token
+  airtable_access_token?: string | null; // Airtable OAuth Access Token (if using OAuth)
   schedule?: ScheduleConfig | null;
 }
 
@@ -41,18 +45,22 @@ interface ScheduleConfig {
   cronExpression?: string;
 }
 
-interface RobotCreationAttributes extends Optional<RobotAttributes, 'id'> { }
+interface RobotCreationAttributes extends Optional<RobotAttributes, 'id'> {}
 
 class Robot extends Model<RobotAttributes, RobotCreationAttributes> implements RobotAttributes {
   public id!: string;
   public userId!: number;
   public recording_meta!: RobotMeta;
   public recording!: RobotWorkflow;
   public google_sheet_email!: string | null;
-  public google_sheet_name?: string | null;
-  public google_sheet_id?: string | null;
+  public google_sheet_name!: string | null;
+  public google_sheet_id!: string | null;
   public google_access_token!: string | null;
   public google_refresh_token!: string | null;
+  public airtable_base_id!: string | null; // Airtable Base ID
+  public airtable_table_name!: string | null; // Airtable Table Name
+  public airtable_personal_access_token!: string | null; // Airtable Personal Access Token
+  public airtable_access_token!: string | null; // Airtable OAuth Access Token
   public schedule!: ScheduleConfig | null;
 }
 
@@ -95,6 +103,22 @@ Robot.init(
       type: DataTypes.STRING,
       allowNull: true,
     },
+    airtable_base_id: {
+      type: DataTypes.STRING,
+      allowNull: true,
+    },
+    airtable_table_name: {
+      type: DataTypes.STRING,
+      allowNull: true,
+    },
+    airtable_personal_access_token: {
+      type: DataTypes.STRING,
+      allowNull: true,
+    },
+    airtable_access_token: {
+      type: DataTypes.STRING,
+      allowNull: true,
+    },
     schedule: {
       type: DataTypes.JSONB,
       allowNull: true,
@@ -107,6 +131,7 @@ Robot.init(
   }
 );
 
+// Uncomment and define relationships if needed
 // Robot.hasMany(Run, {
 //   foreignKey: 'robotId',
 //   as: 'runs', // Alias for the relation

server/src/workflow-management/integrations/airtableintegration.ts

@@ -0,0 +1,139 @@
+import Airtable from 'airtable';
+import logger from '../../logger';
+import Run from '../../models/Run';
+import Robot from '../../models/Robot';
+
+interface AirtableUpdateTask {
+  robotId: string;
+  runId: string;
+  status: 'pending' | 'completed' | 'failed';
+  retries: number;
+}
+
+const MAX_RETRIES = 5;
+
+export let airtableUpdateTasks: { [runId: string]: AirtableUpdateTask } = {};
+
+/**
+ * Updates Airtable with data from a successful run.
+ * @param robotId - The ID of the robot.
+ * @param runId - The ID of the run.
+ */
+export async function updateAirtable(robotId: string, runId: string) {
+  try {
+    const run = await Run.findOne({ where: { runId } });
+
+    if (!run) {
+      throw new Error(`Run not found for runId: ${runId}`);
+    }
+
+    const plainRun = run.toJSON();
+
+    if (plainRun.status === 'success') {
+      let data: { [key: string]: any }[] = [];
+      if (plainRun.serializableOutput && Object.keys(plainRun.serializableOutput).length > 0) {
+        data = plainRun.serializableOutput['item-0'] as { [key: string]: any }[];
+      } else if (plainRun.binaryOutput && plainRun.binaryOutput['item-0']) {
+        const binaryUrl = plainRun.binaryOutput['item-0'] as string;
+        data = [{ "Screenshot URL": binaryUrl }];
+      }
+
+      const robot = await Robot.findOne({ where: { 'recording_meta.id': robotId } });
+
+      if (!robot) {
+        throw new Error(`Robot not found for robotId: ${robotId}`);
+      }
+
+      const plainRobot = robot.toJSON();
+
+      const tableName = plainRobot.airtable_table_name;
+      const baseId = plainRobot.airtable_base_id;
+      const personalAccessToken = plainRobot.airtable_personal_access_token;
+
+      if (tableName && baseId && personalAccessToken) {
+        console.log(`Preparing to write data to Airtable for robot: ${robotId}, table: ${tableName}`);
+
+        await writeDataToAirtable(baseId, tableName, personalAccessToken, data);
+        console.log(`Data written to Airtable successfully for Robot: ${robotId} and Run: ${runId}`);
+      } else {
+        console.log('Airtable integration not configured.');
+      }
+    } else {
+      console.log('Run status is not success or serializableOutput is missing.');
+    }
+  } catch (error: any) {
+    console.error(`Failed to write data to Airtable for Robot: ${robotId} and Run: ${runId}: ${error.message}`);
+  }
+}
+
+/**
+ * Writes data to Airtable.
+ * @param baseId - The ID of the Airtable base.
+ * @param tableName - The name of the Airtable table.
+ * @param personalAccessToken - The Airtable Personal Access Token.
+ * @param data - The data to write to Airtable.
+ */
+export async function writeDataToAirtable(baseId: string, tableName: string, personalAccessToken: string, data: any[]) {
+  try {
+    // Initialize Airtable with Personal Access Token
+    const base = new Airtable({ apiKey: personalAccessToken }).base(baseId);
+
+    const table = base(tableName);
+
+    // Prepare records for Airtable
+    const records = data.map((row) => ({ fields: row }));
+
+    // Write data to Airtable
+    const response = await table.create(records);
+
+    if (response) {
+      console.log('Data successfully appended to Airtable.');
+    } else {
+      console.error('Airtable append failed:', response);
+    }
+
+    logger.log(`info`, `Data written to Airtable: ${tableName}`);
+  } catch (error: any) {
+    logger.log(`error`, `Error writing data to Airtable: ${error.message}`);
+    throw error;
+  }
+}
+
+/**
+ * Processes pending Airtable update tasks.
+ */
+export const processAirtableUpdates = async () => {
+  while (true) {
+    let hasPendingTasks = false;
+    for (const runId in airtableUpdateTasks) {
+      const task = airtableUpdateTasks[runId];
+      console.log(`Processing task for runId: ${runId}, status: ${task.status}`);
+
+      if (task.status === 'pending') {
+        hasPendingTasks = true;
+        try {
+          await updateAirtable(task.robotId, task.runId);
+          console.log(`Successfully updated Airtable for runId: ${runId}`);
+          delete airtableUpdateTasks[runId];
+        } catch (error: any) {
+          console.error(`Failed to update Airtable for run ${task.runId}:`, error);
+          if (task.retries < MAX_RETRIES) {
+            airtableUpdateTasks[runId].retries += 1;
+            console.log(`Retrying task for runId: ${runId}, attempt: ${task.retries}`);
+          } else {
+            airtableUpdateTasks[runId].status = 'failed';
+            console.log(`Max retries reached for runId: ${runId}. Marking task as failed.`);
+          }
+        }
+      }
+    }
+
+    if (!hasPendingTasks) {
+      console.log('No pending tasks. Exiting loop.');
+      break;
+    }
+
+    console.log('Waiting for 5 seconds before checking again...');
+    await new Promise((resolve) => setTimeout(resolve, 5000));
+  }
+};

server/src/workflow-management/scheduler/index.ts

@@ -6,6 +6,7 @@ import { createRemoteBrowserForRun, destroyRemoteBrowser } from '../../browser-m
 import logger from '../../logger';
 import { browserPool } from "../../server";
 import { googleSheetUpdateTasks, processGoogleSheetUpdates } from "../integrations/gsheet";
+import { airtableUpdateTasks, processAirtableUpdates } from "../integrations/airtableintegration"; // Import Airtable functions
 import Robot from "../../models/Robot";
 import Run from "../../models/Run";
 import { getDecryptedProxyConfig } from "../../routes/proxy";
@@ -44,7 +45,7 @@ async function createWorkflowAndStoreMetadata(id: string, userId: string) {
       };
     }
 
-    const browserId = createRemoteBrowserForRun( userId);
+    const browserId = createRemoteBrowserForRun(userId);
     const runId = uuid();
 
     const run = await Run.create({
@@ -177,13 +178,24 @@ async function executeRun(id: string) {
       }
     );
 
+    // Add task for Google Sheets update
     googleSheetUpdateTasks[id] = {
       robotId: plainRun.robotMetaId,
       runId: id,
       status: 'pending',
       retries: 5,
     };
     processGoogleSheetUpdates();
+
+    // Add task for Airtable update
+    airtableUpdateTasks[id] = {
+      robotId: plainRun.robotMetaId,
+      runId: id,
+      status: 'pending',
+      retries: 5,
+    };
+    processAirtableUpdates();
+
     return true;
   } catch (error: any) {
     logger.log('info', `Error while running a robot with id: ${id} - ${error.message}`);