Merge pull request #8 from imarc/feature/CPA-2350-individual-sync

Feature/cpa 2350 individual sync

Author: btam06

src/Mapping.php

@@ -7,6 +7,12 @@
  */
 class Mapping
 {
+	/**
+	 * 
+	 */
+	protected $adjuncts = array();
+
+	
 	/**
 	 *
 	 */
@@ -110,6 +116,17 @@ public function __construct($source, $destination, $key)
 	}
 
 
+	/**
+	 * 
+	 */
+	public function addAdjunct($adjunct, $keys)
+	{
+		$this->adjuncts[$adjunct] = $keys;
+
+		return $this;
+	}
+
+
 	/**
 	 *
 	 */
@@ -256,6 +273,28 @@ public function compose($string, ...$params)
 	}
 
 
+	/**
+	 * 
+	 */
+	public function composeSourceAdjunctKeyQuery(Mapping $adjunct, $ids)
+	{
+		$table = $adjunct->getDestination();
+		$keys  = $this->adjuncts[$table];
+
+		$sql = $this->compose(
+			'SELECT %s FROM %s WHERE %s AND (%s)',
+			$adjunct->makeSourceKey(),
+			$adjunct->makeSourceFrom(),
+			$adjunct->makeSourceWheres(),
+			join(' OR ', array_map(function($id) use ($keys) {
+				return sprintf("RTRIM(LTRIM(%s)) = '%s'", $keys['source'], $id[$this->key[0]]);
+			}, $ids))
+		);
+
+		return $sql;
+	}
+
+
 	/**
 	 *
 	 */
@@ -315,7 +354,7 @@ public function composeSourceInsertSelectQuery()
 	/**
 	 * 
 	 */
-	public function composeSourceDeleteSelectQuery()
+	public function composeSourceDeleteSelectQuery($ids = array())
 	{
 		$keys = [];
 		foreach ($this->getKey() as $key) {
@@ -324,17 +363,25 @@ public function composeSourceDeleteSelectQuery()
 
 		$join_keys = join(' AND ', $keys);
 
-		$sql = $this->compose(
-			'SELECT %s.* from devour_temp_%s RIGHT OUTER JOIN %s ON (%s) WHERE devour_temp_%s.%s IS NULL',
+		$sql = 'SELECT %s.* from devour_temp_%s RIGHT OUTER JOIN %s ON (%s) WHERE devour_temp_%s.%s IS NULL';
+
+		$params = [
 			$this->getDestination(),
 			$this->getDestination(),
 			$this->getDestination(),
 			$join_keys,
 			$this->getDestination(),
 			$this->key[0]
-		);
+		];
 
-		return $sql;
+		if (!empty($ids)) {
+			if ($keys) {
+				$sql .= ' AND %s';
+				$params[] = $this->makeDestinationInKeys($ids);
+			}
+		}
+
+		return $this->compose($sql, ...$params);
 	}
 
 
@@ -392,7 +439,7 @@ public function composeSourceSelectQuery($keys = NULL)
 			$sql .= ' AND %s';
 			$params[] = $this->makeSourceInKeys($keys);
 		}
-		
+
 		$sql = $this->compose(
 			$sql,
 			...$params
@@ -419,6 +466,31 @@ public function composeSourceUpdatedKeysQuery(array $existing_keys)
 	}
 
 
+	/**
+	 * 
+	 */
+	public function composeKeys($ids)
+	{
+		return array_map(function($key) {
+			$keys = [];
+			foreach ($this->key as $id) {
+				$keys[$id] = $key;
+			}
+			
+			return $keys;
+		}, $ids);
+	}
+
+
+	/**
+	 * 
+	 */
+	public function getAdjuncts()
+	{
+		return $this->adjuncts;
+	}
+
+
 	/**
 	 *
 	 */
@@ -544,9 +616,9 @@ protected function makeDestinationInKeys(array $keys)
 
 			return sprintf($group, implode(' AND ', array_map(function($field) use ($key) {
 				if (is_string($key[$field])) {
-					return sprintf("%s = '%s'", $field, str_replace("'", "''", $key[$field]));
+					return sprintf("%s.%s = '%s'", $this->destination, $field, str_replace("'", "''", $key[$field]));
 				} else {
-					return sprintf("%s = %s", $field, $key[$field]);
+					return sprintf("%s.%s = %s", $this->destination, $field, $key[$field]);
 				}
 			}, $this->key)));
 		}, $keys)));
@@ -651,9 +723,9 @@ protected function makeSourceInKeys(array $keys)
 
 			return sprintf($group, implode(' AND ', array_map(function($field) use ($key) {
 				if (is_string($key[$field])) {
-					return sprintf("%s = '%s'", $this->fields[$field], str_replace("'", "''", $key[$field]));
+					return sprintf("RTRIM(LTRIM(%s)) = '%s'", $this->fields[$field], str_replace("'", "''", $key[$field]));
 				} else {
-					return sprintf("%s = %s", $this->fields[$field], $key[$field]);
+					return sprintf("RTRIM(LTRIM(%s)) = %s", $this->fields[$field], $key[$field]);
 				}
 			}, $this->key)));
 		}, $keys)));

src/Synchronizer.php

@@ -361,6 +361,23 @@ public function getLastSyncTime(): ?string
 		return strtotime($result->fetch(PDO::FETCH_ASSOC)['start_time']);
 	}
 
+	/**
+	 * 
+	 */
+	public function getMapping($name)
+	{
+		return $this->mappings[$name] ?? NULL;
+	}
+
+
+	/**
+	 * 
+	 */
+	public function getMappings(): array
+	{
+		return $this->mappings;
+	}
+
 
 	/**
 	 *
@@ -441,7 +458,7 @@ public function schedule(array $mappings = array(), $scheduled_by = NULL): array
 	/**
 	 *
 	 */
-	public function run(array $mappings = array(), $force_update = FALSE): array
+	public function run(array $mappings = array(), $ids = array(), $force_update = FALSE): array
 	{
 		$this->stat();
 
@@ -479,7 +496,7 @@ public function run(array $mappings = array(), $force_update = FALSE): array
 
 			foreach ($mappings as $mapping) {
 				try {
-					$this->syncMapping($mapping, $force_update);
+					$this->syncMapping($mapping, $ids[$mapping] ?? [], $force_update);
 
 				} catch (\Exception $e) {
 					$this->log($e->getMessage());
@@ -848,7 +865,7 @@ protected function log($message)
 	/**
 	 *
 	 */
-	protected function syncMapping($name, $force_update)
+	protected function syncMapping($name, $ids, $force_update)
 	{
 		if (!isset($this->mappings[$name])) {
 			throw new RuntimeException(sprintf(
@@ -875,7 +892,7 @@ protected function syncMapping($name, $force_update)
 		//
 
 		foreach ($mapping->getDependencies() as $dependency) {
-			$this->syncMapping($dependency, $force_update);
+			$this->syncMapping($dependency, [], $force_update);
 		}
 
 		if ($this->strictTime) {
@@ -889,17 +906,17 @@ protected function syncMapping($name, $force_update)
 		$this->log(sprintf('Syncing %s', $name));
 
 		$this->createTemporaryTable($mapping);
-		$this->syncMappingTemporary($mapping);
+		$this->syncMappingTemporary($mapping, $ids);
 
 		$start_sync_time  = date('Y-m-d H:i:s');
 
-		if ($this->truncate[$mapping->getDestination()]) {
+		if ($this->truncate[$mapping->getDestination()] && empty($ids)) {
 			$this->truncateTable($mapping);
 			$this->syncMappingInserts($mapping);
 
 		} else {
 			if ($mapping->canDelete()) {
-				$this->syncMappingDeletes($mapping);
+				$this->syncMappingDeletes($mapping, $ids);
 				$this->log('...completed deletions');
 			}
 
@@ -920,20 +937,32 @@ protected function syncMapping($name, $force_update)
 		$this->updateSet($name, $start_sync_time);
 
 		$this->synced[array_pop($this->stack)] = TRUE;
+
+		if ($ids) {
+			foreach ($mapping->getAdjuncts() as $adjunct => $config) {
+				$adjunct   = $this->mappings[$adjunct];
+				$key_query = $mapping->composeSourceAdjunctKeyQuery($adjunct, $ids);
+				$keys      = $this->source->query($key_query)->fetchAll();
+
+				$keys = $this->filterKeys($adjunct, $keys, 'select');
+
+				$this->syncMapping($adjunct->getDestination(), $keys, $force_update);
+			}
+		}
 	}
 
 
 	/**
 	 *
 	 */
-	protected function syncMappingDeletes(Mapping $mapping)
+	protected function syncMappingDeletes(Mapping $mapping, $ids = array())
 	{
 		if (!$mapping->canDelete()) {
 			return;
 		}
 
 		try {
-			$delete_select_query = $mapping->composeSourceDeleteSelectQuery();
+			$delete_select_query = $mapping->composeSourceDeleteSelectQuery($ids);
 			$delete_results      = $this->destination->query($delete_select_query)->fetchAll();
 		} catch (\Exception $e) {
 			$this->log(sprintf(
@@ -1044,9 +1073,9 @@ protected function syncMappingInserts(Mapping $mapping)
 	/**
 	 *
 	 */
-	protected function syncMappingTemporary(Mapping $mapping)
+	protected function syncMappingTemporary(Mapping $mapping, $ids = array())
 	{
-		$source_select_query = $mapping->composeSourceSelectQuery();
+		$source_select_query = $mapping->composeSourceSelectQuery($ids);
 		try {
 			$source_results = $this->source->query($source_select_query, PDO::FETCH_ASSOC)->fetchAll();
 			$generated      = array_keys($mapping->getGenerators());