Merge development (with our current Datawarehouse code) into the AWS branch.

.github/workflows/publish-postgres.yml

@@ -0,0 +1,33 @@
+# This workflow will do a clean install of node dependencies, cache/restore them, build the source code and run tests across different versions of node
+# For more information see: https://help.github.com/actions/language-and-framework-guides/using-nodejs-with-github-actions
+
+name: Publish-Postgres
+
+on: 
+  release:
+    types: [published]
+    # branches: [ $default-branch ]
+
+jobs:
+  publish-new-release:
+    runs-on: ubuntu-latest
+    defaults:
+      run:
+        working-directory: ./postgres
+    steps:
+    - uses: actions/checkout@v2
+    - name: Node 12
+      uses: actions/setup-node@v2
+      with:
+        node-version: '12.x'
+        registry-url: 'https://npm.pkg.github.com'
+        scope: '@leoplatform'
+    - run: npm ci
+    - run: npm run build --if-present
+    # - run: npm run test #Nothing is implemented here currently so this is omitted
+    # - run: git config user.email "githubactions@commercehub.com" #skipping this for now since we don't have a servie account for LeoPlatform yet
+    - run: git config user.name "GitHub-Actions"
+    - run: npm version ${{ github.event.release.tag_name }}
+    - run: npm publish --access=public 
+      env:
+        NODE_AUTH_TOKEN: ${{ secrets.GITHUB_TOKEN }}

common/datawarehouse/combine.js

@@ -118,7 +118,10 @@ function combine(file) {
 				process.exit();
 			}
 			if (lastObj && id === lastId) {
-				if (data.__leo_delete__) {
+				if (data.__leo_delete__ || lastObj.__leo_delete__) {
+					// if our current row is a delete, then we throw away the previous updates
+					// if our previous row was a delete, but the current one is not, we throw away the delete, because it was updated
+					// after the delete, so it should NOT be deleted.
 					lastObj = data;
 				} else {
 					lastObj = merge(lastObj, data);

common/datawarehouse/load.js

@@ -9,7 +9,7 @@ const async = require('async');
 const validate = require('./../utils/validation');
 let errorStream;
 
-module.exports = function (ID, source, client, tableConfig, stream, callback) {
+module.exports = function(ID, source, client, tableConfig, stream, callback) {
 
 	// adding backwards compatibility for ID and source
 	if (!callback) {
@@ -39,7 +39,7 @@ module.exports = function (ID, source, client, tableConfig, stream, callback) {
 		});
 	});
 
-	let checkforDelete = ls.through(function (obj, done) {
+	let checkforDelete = ls.through(function(obj, done) {
 		if (obj.payload.type === 'delete') {
 			let data = obj.payload.data || {};
 			let ids = data.in || [];
@@ -68,7 +68,7 @@ module.exports = function (ID, source, client, tableConfig, stream, callback) {
 		}
 	});
 
-	let validateData = ls.through(function (obj, done) {
+	let validateData = ls.through(function(obj, done) {
 		let eventObj = obj.payload.data;
 		let invalid = false;
 
@@ -336,7 +336,7 @@ module.exports = function (ID, source, client, tableConfig, stream, callback) {
  * @param eventObj {Object}
  * @param error {string}
  */
-function handleFailedValidation (ID, source, eventObj, error) {
+function handleFailedValidation(ID, source, eventObj, error) {
 	if (!errorStream || !errorStream.Writable) {
 		errorStream = streams.passthrough({
 			objectMode: true
@@ -345,11 +345,11 @@ function handleFailedValidation (ID, source, eventObj, error) {
 		streams.pipe(
 			errorStream,
 
-			ls.process(ID, obj => {
+			ls.process(ID, function(obj) {
 				return obj;
-			}),
-
-			leo.load(ID, `${source}_error`),
+			}, `${source}_error`),
+			ls.toLeo(ID),
+			ls.devnull(),
 
 			(err) => {
 				err && logger.error('GOT ERROR', err);

common/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "leo-connector-common",
-	"version": "4.0.11-rc",
+	"version": "4.0.13-rc",
 	"description": "Common package for all Leo Platform database connectors",
 	"main": "index.js",
 	"directories": {},

entity-table/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "leo-connector-entity-table",
-	"version": "3.0.21-rc",
+	"version": "3.0.22-rc",
 	"description": "A Leo entity table connector",
 	"repository": {
 		"type": "git",

postgres/lib/connect.js

@@ -11,14 +11,16 @@ var copyFrom = require('pg-copy-streams').from;
 var copyTo = require('pg-copy-streams').to;
 let csv = require('fast-csv');
 
+// need to confirm there is no issue adding these dependencies
+const leo = require('leo-sdk');
+const ls = leo.streams;
+
 require('pg').types.setTypeParser(1114, (val) => {
 	val += 'Z';
 	logger.debug(val);
 	return moment(val).unix() + '  ' + moment(val).utc().format();
 });
 
-const ls = require('leo-sdk').streams;
-
 let queryCount = 0;
 module.exports = function (config) {
 	const pool = new Pool(Object.assign({
@@ -571,8 +573,176 @@ function create (pool, parentCache) {
 				records: opts.records,
 			});
 		},
-		streamToTableFromS3: () => {
-			// opts = Object.assign({}, opts || {});
+		streamToTableFromS3: (table, config) => {
+			const ts = table.split('.');
+			let schema = 'public';
+			let shortTable = table;
+			if (ts.length > 1) {
+				schema = ts[0];
+				shortTable = ts[1];
+			}
+
+			let columns = [];
+			let stream;
+			let myClient = null;
+			let pending = null;
+			let ended = false;
+			let csvopts = { delimiter: '|' };
+			let keepS3Files = config.keepS3Files != null ? config.keepS3Files : false;
+
+			// Get prefix for S3 file path
+			let s3prefix =
+					config.s3prefix ||
+					process.env.AWS_LAMBDA_FUNCTION_NAME ||
+					'dw_redshift_ingest';
+			s3prefix = s3prefix.replace(/^\/*(.*?)\/*$/, '$1'); // Remove leading and trailing '/'
+
+			// clean audit date to use in S3 file path
+			let cleanAuditDate = client.auditdate
+				.replace(/'/g, '')
+				.replace(/:/g, '-');
+			let s3FileName = `files/${s3prefix}/${cleanAuditDate}/${table}.csv`;
+
+			client.connect().then(
+				c => {
+					client
+						.describeTable(shortTable, schema)
+						.then(result => {
+							if (ended) {
+								c.release(true);
+								return;
+							}
+							columns = result.map(f => f.column_name);
+							myClient = c;
+
+							stream = ls.toS3(leo.configuration.resources.LeoS3, s3FileName);
+							// copyFrom uses `end` but s3 `finish` so pipe finish to end
+							stream.on('finish', () => stream.emit('end'));
+
+							stream.on('error', function(err) {
+								logger.error(`COPY error: ${err.where}`, err);
+								process.exit();
+							});
+							if (pending) {
+								pending();
+							}
+						})
+						.catch(err => {
+							logger.error(err);
+						});
+				},
+				err => {
+					logger.error(err);
+				}
+			);
+
+			let count = 0;
+
+			function nonNull(v) {
+				if (v === '' || v === null || v === undefined) {
+					return '\\N';
+				} else if (typeof v === 'string' && (v.search(/\r/) !== -1 || v.search(/\n/) !== -1)) {
+					if (config.version !== 'redshift') {
+						return v.replace(/\r\n?/g, '\n');
+					} else {
+						return v.replace(/\r\n?/g, '\n').replace(/\n/g, `\\n`);
+					}
+				} else {
+					return v;
+				}
+			}
+
+
+			return ls.pipeline(
+				csv.createWriteStream({
+					...csvopts,
+					headers: false,
+					transform: (row, done) => {
+						if (!myClient) {
+							pending = () => {
+								done(
+									null,
+									columns.map(f => nonNull(row[f]))
+								);
+							};
+						} else {
+							done(
+								null,
+								columns.map(f => nonNull(row[f]))
+							);
+						}
+					},
+				}),
+				ls.write(
+					(r, done) => {
+						count++;
+						if (count % 10000 === 0) {
+							logger.info(table + ': ' + count);
+						}
+						if (!stream.write(r)) {
+							stream.once('drain', done);
+						} else {
+							done(null);
+						}
+					},
+					done => {
+						ended = true;
+						logger.debug(table + ': stream done');
+						if (stream) {
+							stream.on('end', err => {
+								logger.debug(table + ': stream ended', err || '');
+
+								// wrap done callback to release the connection
+								function innerDone(err) {
+									myClient.release(true);
+									logger.debug(table + ': stream client released', err || '');
+									done(err);
+								}
+
+								if (err) {
+									innerDone(err);
+								} else {
+									// Once the S3 file is complete run copy to load the staging table
+									let f = columns.map(f => `"${f}"`);
+									let file = `s3://${leo.configuration.s3}/${s3FileName}`;
+									let manifest = '';
+									let role = config.loaderRole;
+									myClient.query(
+										`copy ${table} (${f}) from '${file}' ${manifest} ${role ? `credentials 'aws_iam_role=${role}'` : ''
+										} NULL AS '\\\\N' format csv DELIMITER '|' ACCEPTINVCHARS TRUNCATECOLUMNS ACCEPTANYDATE TIMEFORMAT 'auto' COMPUPDATE OFF`,
+										copyErr => {
+											if (keepS3Files) {
+												innerDone(copyErr);
+											} else {
+												// Delete the S3 files when done
+												ls.s3.deleteObject(
+													{
+														Bucket: leo.configuration.s3,
+														Key: s3FileName,
+													},
+													deleteError => {
+														if (deleteError) {
+															logger.info(
+																'file failed to delete:',
+																s3FileName,
+																deleteError
+															);
+														}
+														innerDone(copyErr);
+													}
+												);
+											}
+										}
+									);
+								} 
+							});
+							stream.end();
+						} else {
+							done();
+						}
+					}
+				)
+			);
 		},
 	};