DynamoDB support (WIP)

package.json

@@ -21,6 +21,7 @@
     "test:mongo4": "npm run test:mongo",
     "test:google-sheets": "lerna exec --scope external-db-google-sheets \"npm test\" && lerna exec --scope velo-external-db \"npm run test:google-sheets\"",
     "test:airtable": "lerna exec --scope external-db-airtable \"npm test\" && lerna exec --scope velo-external-db \"npm run test:airtable\"",
+    "test:dynamo": "lerna exec --scope external-db-dynamo \"npm test\" && lerna exec --scope velo-external-db \"npm run test:dynamo\"",
     "release": "lerna version --no-push",
     "outdated:all": "lerna exec --scope '{external-db-config,external-db-firestore,external-db-mongo,external-db-mssql,external-db-mysql,external-db-postgres,external-db-spanner,external-db-google-sheets,external-db-airtable,velo-external-db,velo-external-db-core}' \"npm outdated\"",
     "ci:publish": "lerna publish -y from-git"
@@ -40,6 +41,7 @@
       "external-db-spanner",
       "external-db-google-sheets",
       "external-db-airtable",
+      "external-db-dynamo",
       "velo-external-db",
       "velo-external-db-core"
     ]

packages/external-db-config/lib/factory.js

@@ -14,8 +14,14 @@ const create = () => {
   switch (vendor.toLowerCase()) {
 
     case 'aws':
-      internalConfigReader = new aws.AwsConfigReader(secretId || DefaultSecretId, region)
-      break;
+      switch(type) {
+        case 'dynamo':
+          internalConfigReader = new aws.AwsDynamoConfigReader(region) 
+          break
+        default:
+          internalConfigReader = new aws.AwsConfigReader(secretId || DefaultSecretId, region)
+      }
+      break
 
     case 'gcp':
       switch (type) {
@@ -58,6 +64,9 @@ const create = () => {
         case 'airtable':
           internalConfigReader = new gcp.GcpAirtableConfigReader()
           break;
+        case 'dynamo':
+          internalConfigReader = new aws.AwsDynamoConfigReader(region) 
+          break
         case 'mssql':
         case 'mysql':
         case 'postgres':

packages/external-db-config/lib/readers/aws_config_reader.js

@@ -4,7 +4,7 @@ const { checkRequiredKeys } = require('../utils/config_utils')
 const EmptyAWSConfig = { host: '', username: '', password: '', DB: '', SECRET_KEY: '', error: true }
 
 class AwsConfigReader {
-  constructor (secretId, region) {
+  constructor(secretId, region) {
     this.secretId = secretId
     this.region = region
   }
@@ -35,4 +35,36 @@ class AwsConfigReader {
   }
 }
 
-module.exports = { AwsConfigReader }
+  class AwsDynamoConfigReader {
+    constructor(region) {
+      this.region = region
+     }
+
+     async readExternalConfig() {
+      const client = new SecretsManagerClient({ region: this.region })
+      const data = await client.send(new GetSecretValueCommand({ SecretId: this.secretId }))
+      return JSON.parse(data.SecretString)
+    }
+
+    async readConfig() {
+      if (process.env.NODE_ENV === 'test') {
+        return { region:this.region, secretKey: process.env.SECRET_KEY, endpoint: process.env.ENDPOINT_URL }
+      }
+      const cfg = await this.readExternalConfig()
+                            .catch(e => {
+                              console.log(e)
+                              return EmptyAWSConfig
+                            })
+
+      const { SECRET_KEY } = cfg
+      return { region:this.region, secretKey: SECRET_KEY }
+    }
+
+    validate() {
+      return {
+        missingRequiredSecretsKeys: checkRequiredKeys(process.env, ['SECRET_KEY']) 
+      }
+    }
+  }
+
+module.exports = { AwsConfigReader, AwsDynamoConfigReader }

packages/external-db-dynamo/.gitignore

@@ -0,0 +1,7 @@
+package-lock.json
+node_modules
+config.json
+schemas.json
+.gcloudignore
+.eslint
+.idea

packages/external-db-dynamo/jest.config.js

@@ -0,0 +1,8 @@
+module.exports = {
+    clearMocks: true,
+    verbose: true,
+    roots: ['<rootDir>/lib'],
+    preset: "ts-jest",
+    testRegex: '(.*\\.spec\\.)js$',
+    testEnvironment: "node"
+};

packages/external-db-dynamo/lib/connection_provider.js

@@ -0,0 +1,25 @@
+const SchemaProvider = require('./dynamo_schema_provider')
+const DataProvider  = require('./dynamo_data_provider')
+const FilterParser = require('./sql_filter_transformer')
+const DatabaseOperations = require('./dynamo_operations')
+const { DynamoDB } = require('@aws-sdk/client-dynamodb')
+
+const extraOptions  = (cfg) => { 
+    if (cfg.endpoint)
+        return { endpoint: cfg.endpoint }
+}
+
+const init = async(cfg, _cfgOptions) => { //remove async
+    //cfg : region , endpoint(optional) ,  
+    const options = _cfgOptions || {}
+    const client = new DynamoDB({region:cfg.region, ...extraOptions(cfg), ...options})
+    const databaseOperations = new DatabaseOperations(client)
+
+    const filterParser = new FilterParser()
+    const dataProvider = new DataProvider(client, filterParser)
+    const schemaProvider = new SchemaProvider(client)
+
+    return { dataProvider, schemaProvider, databaseOperations, connection: client, cleanup: () => {}}
+}
+
+module.exports = init

packages/external-db-dynamo/lib/dynamo_data_provider.js

@@ -0,0 +1,123 @@
+const { patchDateTime, updateFieldsFor } = require('velo-external-db-commons')
+const { DynamoDBDocument }  = require ('@aws-sdk/lib-dynamodb')
+const { validateTable, patchFixDates } = require('./dynamo_utils')
+
+class DataProvider {
+    constructor(client, filterParser) {
+        this.filterParser = filterParser
+        this.client = client
+        this.docClient = DynamoDBDocument.from(client)
+
+    }
+
+    async find(collectionName, filter, sort, skip, limit) {
+        const {filterExpr} = this.filterParser.transform(filter)
+        const response = await this.docClient
+                                   .scan({TableName: collectionName, 
+                                          ...filterExpr,
+                                          Limit:limit
+                                          })
+        return response.Items.map(patchFixDates)
+    }
+
+    async count(collectionName, filter) {
+        const {filterExpr} = this.filterParser.transform(filter)
+        const response = await this.docClient
+                         .scan({TableName: collectionName, 
+                                ...filterExpr,
+                                Select: 'COUNT'
+                                })
+        return response.Count
+    }
+
+    async insert(collectionName, items) {
+        validateTable()
+        await this.docClient
+                  .batchWrite(this.batchPutItemsExpression(collectionName, items.map(patchDateTime)))
+        return items.length //check if there is a way to figure how many deleted/inserted with batchWrite
+    }
+
+    async update(collectionName, items) {
+        await this.docClient.transactWrite({
+            TransactItems:items.map(item=>this.updateSingleItemExpression(collectionName, patchDateTime(item)))
+        })
+        return items.length
+    }
+
+    async delete(collectionName, itemIds) {
+        validateTable()
+        await this.docClient
+                  .batchWrite(this.batchDeleteItemsExpression(collectionName, itemIds))
+        return itemIds.length 
+    }
+
+    async truncate(collectionName) {
+        validateTable(collectionName)
+        const rows = await this.docClient
+                               .scan({
+                                    TableName: collectionName,
+                                    AttributesToGet: ['_id']
+                                })
+
+        await this.docClient
+                  .batchWrite(this.batchDeleteItemsExpression(collectionName, rows.Items.map(item=>item._id)))
+    }
+
+    batchPutItemsExpression(collectionName, items) {
+        return {
+            RequestItems: {
+                [collectionName]: items.map(this.putSingleItemExpression)
+              }
+        }
+    }
+
+    putSingleItemExpression(item) {
+        return {
+            PutRequest: {
+                Item: {
+                    ...item
+                }
+            }
+        }
+    }
+
+    batchDeleteItemsExpression(collectionName, itemIds) {
+        return {
+            RequestItems: {
+                [collectionName]: itemIds.map(this.deleteSingleItemExpression)
+              }
+        }
+    }
+
+    deleteSingleItemExpression(id) {
+        return {
+            DeleteRequest: {
+                Key: {
+                    _id : id
+                }
+            }
+        }
+    }
+
+    updateSingleItemExpression(collectionName, item) {
+        const updateFields = updateFieldsFor(item)
+        const updateExpression = `SET ${updateFields.map(f => `#${f} = :${f}`).join(', ')}`
+        const expressionAttributeNames = updateFields.reduce((pv, cv)=> ({ ...pv, [`#${cv}`]: cv }), {})
+        const expressionAttributeValues = updateFields.reduce((pv, cv)=> ({ ...pv, [`:${cv}`]: item[cv]}), {})
+
+        return {
+            Update: {
+                TableName: collectionName,
+                Key: {
+                    _id: item._id
+                },
+                UpdateExpression: updateExpression,
+                ExpressionAttributeNames: expressionAttributeNames,
+                ExpressionAttributeValues: expressionAttributeValues
+            }
+        }
+    }
+
+}
+
+module.exports = DataProvider

packages/external-db-dynamo/lib/dynamo_operations.js

@@ -0,0 +1,14 @@
+const { notThrowingTranslateErrorCodes } = require('./sql_exception_translator')
+
+class DatabaseOperations {
+    constructor(client) {
+        this.client = client
+    }
+
+    async validateConnection() {
+        return await this.client.listTables({}).then(() => { return { valid: true } })
+                                               .catch((e) => { return { valid: false, error: notThrowingTranslateErrorCodes(e) } })
+    }
+}
+
+module.exports = DatabaseOperations