feat(rds/dataloader): Add support for using RDS as a dataloader (#110)

alichherawalla · web-flow · commit 23817b5b0dc0 · 2021-08-11T11:22:10.000+02:00
diff --git a/README.md b/README.md
@@ -67,6 +67,12 @@ Put options under `custom.appsync-simulator` in your `serverless.yml` file
 | dynamoDb.secretAccessKey | DEFAULT_SECRET             | AWS Secret Key to access DynamoDB                                                                                                                                                                    |
 | dynamoDb.sessionToken    | DEFAULT_ACCESS_TOKEEN      | AWS Session Token to access DynamoDB, only if you have temporary security credentials configured on AWS                                                                                              |
 | dynamoDb.\*              |                            | You can add every configuration accepted by [DynamoDB SDK](https://docs.aws.amazon.com/AWSJavaScriptSDK/latest/AWS/DynamoDB.html#constructor-property)                                               |
+| rds.dbName               |                            | Name of the database                                                                                                                                                                                 |
+| rds.dbHost               |                            | Database host                                                                                                                                                                                        |
+| rds.dbDialect            |                            | Database dialect. Possible values (mysql|postgres)                                                                                                                                                   |
+| rds.dbUsername           |                            | Database username                                                                                                                                                                                    |
+| rds.dbPassword           |                            | Database password                                                                                                                                                                                    |
+| rds.dbPort               |                            | Database port                                                                                                                                                                                        |
 | watch                    | - \*.graphql<br/> - \*.vtl | Array of glob patterns to watch for hot-reloading.                                                                                                                                                   |
 
 Example:
@@ -250,10 +256,115 @@ This plugin supports resolvers implemented by `amplify-appsync-simulator`, as we
 
 - AMAZON_ELASTIC_SEARCH
 - HTTP
+- RELATIONAL_DATABASE
 
-**Not Supported / TODO**
+## Relational Database
+### Sample VTL for a create mutation
 
-- RELATIONAL_DATABASE
+```
+#set( $cols = [] )
+#set( $vals = [] )
+#foreach( $entry in $ctx.args.input.keySet() )
+  #set( $regex = "([a-z])([A-Z]+)")
+  #set( $replacement = "$1_$2")
+  #set( $toSnake = $entry.replaceAll($regex, $replacement).toLowerCase() )
+  #set( $discard = $cols.add("$toSnake") )
+  #if( $util.isBoolean($ctx.args.input[$entry]) )
+      #if( $ctx.args.input[$entry] )
+        #set( $discard = $vals.add("1") )
+      #else
+        #set( $discard = $vals.add("0") )
+      #end
+  #else
+      #set( $discard = $vals.add("'$ctx.args.input[$entry]'") )
+  #end
+#end
+#set( $valStr = $vals.toString().replace("[","(").replace("]",")") )
+#set( $colStr = $cols.toString().replace("[","(").replace("]",")") )
+#if ( $valStr.substring(0, 1) != '(' )
+  #set( $valStr = "($valStr)" )
+#end
+#if ( $colStr.substring(0, 1) != '(' )
+  #set( $colStr = "($colStr)" )
+#end
+{
+  "version": "2018-05-29",
+  "statements":   ["INSERT INTO <name-of-table> $colStr VALUES $valStr", "SELECT * FROM    <name-of-table> ORDER BY id DESC LIMIT 1"]
+}
+```
+
+### Sample VTL for an update mutation
+```
+#set( $update = "" )
+#set( $equals = "=" )
+#foreach( $entry in $ctx.args.input.keySet() )
+  #set( $cur = $ctx.args.input[$entry] )
+  #set( $regex = "([a-z])([A-Z]+)")
+  #set( $replacement = "$1_$2")
+  #set( $toSnake = $entry.replaceAll($regex, $replacement).toLowerCase() )
+  #if( $util.isBoolean($cur) )
+      #if( $cur )
+        #set ( $cur = "1" )
+      #else
+        #set ( $cur = "0" )
+      #end
+  #end
+  #if ( $util.isNullOrEmpty($update) )
+      #set($update = "$toSnake$equals'$cur'" )
+  #else
+      #set($update = "$update,$toSnake$equals'$cur'" )
+  #end
+#end
+{
+  "version": "2018-05-29",
+  "statements":   ["UPDATE <name-of-table> SET $update WHERE id=$ctx.args.input.id", "SELECT * FROM <name-of-table> WHERE id=$ctx.args.input.id"]
+}
+```
+
+
+### Sample resolver for delete mutation
+```
+{
+  "version": "2018-05-29",
+  "statements":   ["UPDATE <name-of-table> set deleted_at=NOW() WHERE id=$ctx.args.id", "SELECT * FROM <name-of-table> WHERE id=$ctx.args.id"]
+}
+```
+
+### Sample mutation response VTL with support for handling AWSDateTime
+```
+#set ( $index = -1)
+#set ( $result = $util.parseJson($ctx.result) )
+#set ( $meta = $result.sqlStatementResults[1].columnMetadata)
+#foreach ($column in $meta)
+    #set ($index = $index + 1)
+    #if ( $column["typeName"] == "timestamptz" )
+        #set ($time = $result["sqlStatementResults"][1]["records"][0][$index]["stringValue"] )
+        #set ( $nowEpochMillis = $util.time.parseFormattedToEpochMilliSeconds("$time.substring(0,19)+0000", "yyyy-MM-dd HH:mm:ssZ") )
+        #set ( $isoDateTime = $util.time.epochMilliSecondsToISO8601($nowEpochMillis) )
+        $util.qr( $result["sqlStatementResults"][1]["records"][0][$index].put("stringValue", "$isoDateTime") )
+    #end   
+#end
+#set ( $res = $util.parseJson($util.rds.toJsonString($util.toJson($result)))[1][0] )
+#set ( $response = {} )
+#foreach($mapKey in $res.keySet())
+    #set ( $s = $mapKey.split("_") )
+    #set ( $camelCase="" )
+    #set ( $isFirst=true )
+    #foreach($entry in $s)
+        #if ( $isFirst )
+          #set ( $first = $entry.substring(0,1) )
+        #else
+          #set ( $first = $entry.substring(0,1).toUpperCase() )
+        #end
+        #set ( $isFirst=false )
+        #set ( $stringLength = $entry.length() )
+        #set ( $remaining = $entry.substring(1, $stringLength) )
+        #set ( $camelCase = "$camelCase$first$remaining" )
+    #end
+    $util.qr( $response.put("$camelCase", $res[$mapKey]) )
+#end
+$utils.toJson($response)
+```
 
 ## Contributors ✨
 
diff --git a/package.json b/package.json
@@ -21,18 +21,21 @@
     "/lib"
   ],
   "dependencies": {
-    "amplify-appsync-simulator": "^1.23.9",
+    "amplify-appsync-simulator": "^1.27.4",
     "amplify-nodejs-function-runtime-provider": "^1.1.6",
     "aws-sdk": "^2.792.0",
     "axios": "^0.21.0",
     "babel-jest": "^26.6.3",
+    "bluebird": "^3.7.2",
     "cfn-resolver-lib": "^1.1.7",
     "dataloader": "^2.0.0",
     "fb-watchman": "^2.0.1",
     "globby": "^11.0.3",
     "jest": "^26.6.3",
     "lodash": "^4.17.20",
-    "merge-graphql-schemas": "^1.5.8"
+    "merge-graphql-schemas": "^1.5.8",
+    "mysql2": "^2.2.5",
+    "pg": "^8.6.0"
   },
   "devDependencies": {
     "@babel/cli": "^7.12.1",
diff --git a/src/data-loaders/RelationalDataLoader.js b/src/data-loaders/RelationalDataLoader.js
@@ -0,0 +1,179 @@
+import Promise from 'bluebird';
+import { Client, types as pgTypes } from 'pg';
+import mysql from 'mysql2/promise';
+import { Types } from 'mysql2';
+const FLAGS = {
+  NOT_NULL: 1,
+  PRI_KEY: 2,
+  UNIQUE_KEY: 4,
+  MULTIPLE_KEY: 8,
+  BLOB: 16,
+  UNSIGNED: 32,
+  ZEROFILL: 64,
+  BINARY: 128,
+  ENUM: 256,
+  AUTO_INCREMENT: 512,
+  TIMESTAMP: 1024,
+  SET: 2048,
+  NO_DEFAULT_VALUE: 4096,
+  ON_UPDATE_NOW: 8192,
+  NUM: 32768,
+};
+
+const decToBin = (dec) => parseInt((dec >>> 0).toString(2), 2);
+
+const convertMySQLResponseToColumnMetaData = (rows) => {
+  return rows.map((row) => {
+    // @TODO: Add for the following fields
+    // arrayBaseColumnType,
+    // isCaseSensitive,
+    // isCurrency,
+    // currency,
+    // precision,
+    // scale,
+    // schemaName,
+    return {
+      isAutoIncrement:
+        decToBin(row.flags & FLAGS.AUTO_INCREMENT) === FLAGS.AUTO_INCREMENT,
+      label: row.name,
+      name: row.name,
+      nullable: decToBin(row.flags && FLAGS.NOT_NULL) !== FLAGS.NOT_NULL,
+      type: row.columnType,
+      typeName: Object.keys(Types)
+        .find((key) => Types[key] === row.columnType)
+        .toUpperCase(),
+      isSigned: decToBin(row.flags & FLAGS.UNSIGNED) !== FLAGS.UNSIGNED,
+      autoIncrement:
+        decToBin(row.flags & FLAGS.AUTO_INCREMENT) === FLAGS.AUTO_INCREMENT,
+      tableName: row._buf
+        .slice(row._tableStart, row._tableStart + row._tableLength)
+        .toString(),
+    };
+  });
+};
+const convertSQLResponseToRDSRecords = (rows) => {
+  const records = [];
+
+  rows.forEach((dbObject) => {
+    const record = [];
+    Object.keys(dbObject).forEach((key) => {
+      record.push(
+        dbObject[key] === null
+          ? { isNull: true, null: true }
+          : typeof dbObject[key] === 'string'
+          ? { stringValue: dbObject[key] }
+          : typeof dbObject[key] === 'number'
+          ? { longValue: dbObject[key] }
+          : { stringValue: dbObject[key] },
+      );
+    });
+    records.push(record);
+  });
+  return records;
+};
+
+const convertPostgresSQLResponseToColumnMetaData = (rows) => {
+  return rows.map((row) => {
+    const typeName = Object.keys(pgTypes.builtins)
+      .find((d) => pgTypes.builtins[d] === row.dataTypeID)
+      .toUpperCase();
+    // @TODO: Add support for the following fields
+    // isAutoIncrement,
+    // nullable,
+    // isSigned,
+    // autoIncrement,
+    // tableName,
+    // arrayBaseColumnType,
+    // isCaseSensitive,
+    // isCurrency,
+    // currency,
+    // precision,
+    // scale,
+    // schemaName,
+    return {
+      label: row.name,
+      name: row.name,
+      type: row.dataTypeID,
+      typeName,
+    };
+  });
+};
+const executeSqlStatements = async (client, req) =>
+  Promise.mapSeries(req.statements, (statement) => client.query(statement));
+export default class RelationalDataLoader {
+  constructor(config) {
+    this.config = config;
+  }
+
+  async load(req) {
+    try {
+      const requiredKeys = [
+        'dbDialect',
+        'dbUsername',
+        'dbPassword',
+        'dbHost',
+        'dbName',
+        'dbPort',
+      ];
+      if (!this.config.rds) {
+        throw new Error('RDS configuration not passed');
+      }
+      const missingKey = requiredKeys.find((key) => {
+        return !this.config.rds[key];
+      });
+      if (missingKey) {
+        throw new Error(`${missingKey} is required.`);
+      }
+
+      const dbConfig = {
+        host: this.config.rds.dbHost,
+        user: this.config.rds.dbUsername,
+        password: this.config.rds.dbPassword,
+        database: this.config.rds.dbName,
+        port: this.config.rds.dbPort,
+      };
+
+      const res = {};
+      if (this.config.rds.dbDialect === 'mysql') {
+        const client = await mysql.createConnection(dbConfig);
+        const results = await executeSqlStatements(client, req);
+
+        res.sqlStatementResults = results.map((result) => {
+          if (result.length < 2) {
+            return {};
+          }
+          if (!result[1]) {
+            // not a select query
+            return {
+              numberOfRecordsUpdated: result[0].affectedRows,
+              generatedFields: [],
+            };
+          }
+          return {
+            numberOfRecordsUpdated: result[0].length,
+            records: convertSQLResponseToRDSRecords(result[0]),
+            columnMetadata: convertMySQLResponseToColumnMetaData(result[1]),
+          };
+        });
+      } else if (this.config.rds.dbDialect === 'postgres') {
+        const client = new Client(dbConfig);
+        await client.connect();
+        const results = await executeSqlStatements(client, req);
+        res.sqlStatementResults = results.map((result) => {
+          return {
+            numberOfRecordsUpdated: result.rowCount,
+            records: convertSQLResponseToRDSRecords(result.rows),
+            columnMetadata: convertPostgresSQLResponseToColumnMetaData(
+              result.fields,
+            ),
+            generatedFields: [],
+          };
+        });
+      }
+      return JSON.stringify(res);
+    } catch (e) {
+      console.log(e);
+      return e;
+    }
+  }
+}
diff --git a/src/getAppSyncConfig.js b/src/getAppSyncConfig.js
@@ -82,6 +82,12 @@ export default function getAppSyncConfig(context, appSyncConfig) {
           },
         };
       }
+      case 'RELATIONAL_DATABASE': {
+        return {
+          ...dataSource,
+          rds: context.options.rds,
+        };
+      }
       case 'AWS_LAMBDA': {
         const { functionName } = source.config;
         if (functionName === undefined) {
diff --git a/src/index.js b/src/index.js
@@ -9,6 +9,7 @@ import getAppSyncConfig from './getAppSyncConfig';
 import NotImplementedDataLoader from './data-loaders/NotImplementedDataLoader';
 import ElasticDataLoader from './data-loaders/ElasticDataLoader';
 import HttpDataLoader from './data-loaders/HttpDataLoader';
+import RelationalDataLoader from './data-loaders/RelationalDataLoader';
 import watchman from 'fb-watchman';
 
 const resolverPathMap = {
@@ -27,7 +28,7 @@ class ServerlessAppSyncSimulator {
 
     addDataLoader('HTTP', HttpDataLoader);
     addDataLoader('AMAZON_ELASTICSEARCH', ElasticDataLoader);
-    addDataLoader('RELATIONAL_DATABASE', NotImplementedDataLoader);
+    addDataLoader('RELATIONAL_DATABASE', RelationalDataLoader);
 
     this.hooks = {
       'before:offline:start:init': this.startServers.bind(this),
@@ -268,6 +269,7 @@ class ServerlessAppSyncSimulator {
         refMap: {},
         getAttMap: {},
         importValueMap: {},
+        rds: {},
         dynamoDb: {
           endpoint: `http://localhost:${get(
             this.serverless.service,
diff --git a/yarn.lock b/yarn.lock
