codeheroesdev
diff --git a/‎manifest.json‎
Lines changed: 6 additions & 0 deletions b/‎manifest.json‎
Lines changed: 6 additions & 0 deletions
diff --git a/‎server/adapters/postgres.js‎
Lines changed: 4 additions & 4 deletions b/‎server/adapters/postgres.js‎
Lines changed: 4 additions & 4 deletions
diff --git a/‎server/sync-agent.js‎
Lines changed: 32 additions & 9 deletions b/‎server/sync-agent.js‎
Lines changed: 32 additions & 9 deletions
diff --git a/‎server/utils/transform.js‎
Lines changed: 0 additions & 26 deletions b/‎server/utils/transform.js‎
Lines changed: 0 additions & 26 deletions
@@ -47,6 +47,12 @@
       "type": "string",
       "format": "hidden"
     },
+    {
+      "name": "last_updated_at",
+      "title": "Last updated_at",
+      "type": "string",
+      "format": "hidden"
+    },
     {
       "name": "last_job_id",
       "title": "Last job ID",
 
@@ -41,19 +41,19 @@ export function closeConnection(client) {
  *
  * Params:
  *   @query String*
- *   @last_sync_at String
+ *   @last_updated_at String
  *
  * Return:
  *   @wrappedQuery String
  */
 
-export function wrapQuery(query, last_sync_at) {
+export function wrapQuery(query, last_updated_at) {
   // Wrap the query.
   const wrappedQuery = `WITH __qry__ AS (${query}) SELECT * FROM __qry__`;
 
   // Add a condition if needed.
-  if (query.match(/updated_at/) && last_sync_at) {
-    return `${wrappedQuery} WHERE updated_at >= '${last_sync_at}'`;
+  if (query.match(/updated_at/) && last_updated_at) {
+    return `${wrappedQuery} WHERE updated_at > '${last_updated_at}'`;
   }
   return wrappedQuery;
 }
 
@@ -5,7 +5,6 @@
 const _ = require("lodash");
 
 // Map each record of the stream.
-const transform = require("./utils/transform");
 const map = require("through2-map");
 
 /**
@@ -26,7 +25,6 @@ const awsS3 = require("s3-upload-stream")(awsAccount);
 
 import * as Adapters from "./adapters";
 
-
 /**
  * Export the sync agent for the SQL ship.
  */
@@ -98,14 +96,13 @@ export default class SyncAgent {
   }
 
   streamQuery(query, options = {}) {
-    const { last_sync_at } = options;
+    const { last_updated_at } = options;
 
     // Wrap the query.
-    const wrappedQuery = this.adapter.wrapQuery(query, last_sync_at);
-
+    const wrappedQuery = this.adapter.wrapQuery(query, last_updated_at);
     // Run the method for the specific adapter.
     return this.adapter.streamQuery(this.client, wrappedQuery).then(stream => {
-      stream.on("error", err => this.hull.logger.error("Query error", { message: err.toString() }));
+      stream.on("error", err => this.hull.logger.error("sync.error", { message: err.toString() }));
       return stream;
     });
   }
@@ -127,17 +124,43 @@ export default class SyncAgent {
   startSync(stream, started_sync_at) {
     this.hull.logger.info("sync.start");
     let processed = 0;
-    const progress = map({ objectMode: true }, (user) => {
+    let last_updated_at;
+
+
+    const transform = map({ objectMode: true }, (record) => {
+      const user = {};
       processed += 1;
-      if (processed % 100 === 0) this.hull.logger.info("sync.progress", { processed, elapsed: new Date() - started_sync_at });
+
+      if (processed % 1000 === 0) {
+        this.hull.logger.info("sync.progress", { processed, elapsed: new Date() - started_sync_at });
+      }
+
+      // Add the user id if exists.
+      if (record.external_id) {
+        user.userId = record.external_id.toString();
+      }
+
+      // console.warn("Hello record", { record });
+
+      if (record.updated_at) {
+        last_updated_at = last_updated_at || record.updated_at;
+        if (record.updated_at > last_updated_at) {
+          last_updated_at = record.updated_at;
+        }
+      }
+
+      // Register eveything else inside the "traits" object.
+      user.traits = _.omit(record, "external_id", "updated_at");
+
       return `${JSON.stringify(user)}\n`;
     });
 
-    return this.uploadStream(stream.pipe(transform()).pipe(progress), started_sync_at)
+    return this.uploadStream(stream.pipe(transform), started_sync_at)
       .then(url => this.startImportJob(url))
       .then(job => {
         return this.updateShipSettings({
           last_sync_at: started_sync_at,
+          last_updated_at: last_updated_at || started_sync_at,
           last_job_id: job.id
         });
       })
Original file line number	Diff line number	Diff line change
`@@ -41,19 +41,19 @@ export function closeConnection(client) {`
`41`	`41`	`*`
`42`	`42`	`* Params:`
`43`	`43`	`* @query String*`
`44`		`- * @last_sync_at String`
	`44`	`+ * @last_updated_at String`
`45`	`45`	`*`
`46`	`46`	`* Return:`
`47`	`47`	`* @wrappedQuery String`
`48`	`48`	`*/`
`49`	`49`
`50`		`-export function wrapQuery(query, last_sync_at) {`
	`50`	`+export function wrapQuery(query, last_updated_at) {`
`51`	`51`	`// Wrap the query.`
`52`	`52`	const wrappedQuery = `WITH __qry__ AS (${query}) SELECT * FROM __qry__`;
`53`	`53`
`54`	`54`	`// Add a condition if needed.`
`55`		`- if (query.match(/updated_at/) && last_sync_at) {`
`56`		- return `${wrappedQuery} WHERE updated_at >= '${last_sync_at}'`;
	`55`	`+ if (query.match(/updated_at/) && last_updated_at) {`
	`56`	+ return `${wrappedQuery} WHERE updated_at > '${last_updated_at}'`;
`57`	`57`	`}`
`58`	`58`	`return wrappedQuery;`
`59`	`59`	`}`