GOOWOO-786: MAPI product sync

src/API/Google/Mapi/MerchantApiClient.php

@@ -22,6 +22,8 @@
  */
 class MerchantApiClient {
 
+	private const BATCH_BOUNDARY_PREFIX = 'gla_mapi_batch_boundary';
+
 	/** @var ClientInterface */
 	protected $http;
 
@@ -146,6 +148,130 @@ public function delete( string $path ): array {
 		return $this->request( 'DELETE', $path );
 	}
 
+	/**
+	 * Send a batch of sub-requests as a single multipart/mixed HTTP request and return
+	 * the per-sub-request results keyed by the input key.
+	 *
+	 * @param array<int|string, array{method: string, path: string, body?: array}> $requests
+	 *
+	 * @return array<int|string, array{status: int, body: array}>
+	 * @throws MerchantApiException On a non-2xx response for the batch request itself.
+	 */
+	public function batch( array $requests ): array {
+		return $this->batch_async( $requests )->wait();
+	}
+
+	/**
+	 * Asynchronous variant of {@see batch()}.
+	 *
+	 * @param array<int|string, array{method: string, path: string, body?: array}> $requests
+	 *
+	 * @return PromiseInterface Resolves to array<int|string, array{status: int, body: array}>.
+	 */
+	public function batch_async( array $requests ): PromiseInterface {
+		$request      = $this->build_batch_request( $requests );
+		$method_label = __METHOD__;
+
+		return $this->http->sendAsync( $request )->then(
+			function ( ResponseInterface $response ): array {
+				return $this->parse_batch_response( $response );
+			},
+			function ( $reason ) use ( $method_label ) {
+				if ( $reason instanceof RequestException && $reason->hasResponse() ) {
+					throw new MerchantApiException(
+						$reason->getResponse()->getStatusCode(),
+						$this->decode_response( $reason->getResponse() ),
+						$method_label,
+						$reason
+					);
+				}
+
+				throw $reason;
+			}
+		);
+	}
+
+	/**
+	 * Build the multipart/mixed batch request. Each sub-request is an application/http
+	 * part tagged with a Content-ID so response parts can be mapped back by key.
+	 *
+	 * @param array<int|string, array{method: string, path: string, body?: array}> $requests
+	 *
+	 * @return Request
+	 */
+	protected function build_batch_request( array $requests ): Request {
+		$boundary = self::BATCH_BOUNDARY_PREFIX . '_' . wp_generate_password( 16, false );
+		$body     = '';
+
+		foreach ( $requests as $key => $sub ) {
+			$body .= "--{$boundary}\r\n";
+			$body .= "Content-Type: application/http\r\n";
+			$body .= "Content-ID: <item{$key}>\r\n\r\n";
+			$body .= sprintf( "%s /%s HTTP/1.1\r\n", $sub['method'], ltrim( $sub['path'], '/' ) );
+
+			if ( isset( $sub['body'] ) ) {
+				$body .= "Content-Type: application/json\r\n\r\n";
+				$body .= (string) wp_json_encode( $sub['body'] ) . "\r\n";
+			} else {
+				$body .= "\r\n";
+			}
+		}
+
+		$body .= "--{$boundary}--\r\n";
+
+		return new Request(
+			'POST',
+			$this->base_url . 'batch',
+			[ 'Content-Type' => "multipart/mixed; boundary={$boundary}" ],
+			$body
+		);
+	}
+
+	/**
+	 * Parse a multipart/mixed batch response into per-sub-request results keyed by the
+	 * original request key, extracted from each part's Content-ID.
+	 *
+	 * @param ResponseInterface $response
+	 *
+	 * @return array<int|string, array{status: int, body: array}>
+	 */
+	protected function parse_batch_response( ResponseInterface $response ): array {
+		if ( ! preg_match( '/boundary=(?:"([^"]+)"|([^;\s]+))/', $response->getHeaderLine( 'Content-Type' ), $matches ) ) {
+			do_action( 'woocommerce_gla_error', 'Merchant API batch response had no parseable multipart boundary.', __METHOD__ );
+			return [];
+		}
+
+		$boundary = '' !== ( $matches[1] ?? '' ) ? $matches[1] : $matches[2];
+		$results  = [];
+
+		foreach ( explode( '--' . $boundary, (string) $response->getBody() ) as $part ) {
+			$part = trim( $part );
+			if ( '' === $part || '--' === $part ) {
+				continue;
+			}
+
+			if ( ! preg_match( '/Content-ID:\s*<response-item([^>]+)>/i', $part, $id_match )
+				|| ! preg_match( '#HTTP/\d(?:\.\d)?\s+(\d+)#', $part, $status_match ) ) {
+				continue;
+			}
+
+			$http_offset = strpos( $part, 'HTTP/' );
+			$inner_body  = '';
+			if ( false !== $http_offset && preg_match( '/\r?\n\r?\n(.*)$/s', substr( $part, $http_offset ), $body_match ) ) {
+				$inner_body = trim( $body_match[1] );
+			}
+
+			$decoded = '' === $inner_body ? [] : json_decode( $inner_body, true );
+
+			$results[ $id_match[1] ] = [
+				'status' => (int) $status_match[1],
+				'body'   => is_array( $decoded ) ? $decoded : [],
+			];
+		}
+
+		return $results;
+	}
+
 	/**
 	 * Decode a JSON response body to an array.
 	 *

src/API/Google/Mapi/Services/MapiProductInputsService.php

@@ -89,34 +89,21 @@ public function insert_many( array $inputs, int $concurrency = 10 ): array {
 			$paths_by_index[ $index ] = $this->build_path( $data_source );
 		}
 
-		$client = $this->client;
-
-		$requests = function () use ( $inputs, $client, $paths_by_index ) {
-			foreach ( $inputs as $index => $input ) {
-				yield $index => $client->request_async( 'POST', $paths_by_index[ $index ], $input->to_array() );
+		return $this->run_in_batches(
+			$inputs,
+			$concurrency,
+			__METHOD__,
+			function ( int $index, ProductInput $input ) use ( $paths_by_index ): array {
+				return [
+					'method' => 'POST',
+					'path'   => $paths_by_index[ $index ],
+					'body'   => $input->to_array(),
+				];
+			},
+			function ( array $body ): ProductInput {
+				return ProductInput::from_array( $body );
 			}
-		};
-
-		$successes = [];
-		$failures  = [];
-
-		( new EachPromise(
-			$requests(),
-			[
-				'concurrency' => $concurrency,
-				'fulfilled'   => function ( array $body, int $index ) use ( &$successes ) {
-					$successes[ $index ] = ProductInput::from_array( $body );
-				},
-				'rejected'    => function ( $reason, int $index ) use ( &$failures ) {
-					$failures[ $index ] = $reason;
-				},
-			]
-		) )->promise()->wait();
-
-		return [
-			'successes' => $successes,
-			'failures'  => $failures,
-		];
+		);
 	}
 
 	/**
@@ -245,26 +232,86 @@ public function delete_many( array $inputs, int $concurrency = 10 ): array {
 			$paths_by_index[ $index ] = $this->build_delete_path( $input, $data_source );
 		}
 
-		$client = $this->client;
-
-		$requests = function () use ( $inputs, $client, $paths_by_index ) {
-			foreach ( $inputs as $index => $input ) {
-				yield $index => $client->request_async( 'DELETE', $paths_by_index[ $index ] );
+		return $this->run_in_batches(
+			$inputs,
+			$concurrency,
+			__METHOD__,
+			function ( int $index ) use ( $paths_by_index ): array {
+				return [
+					'method' => 'DELETE',
+					'path'   => $paths_by_index[ $index ],
+				];
+			},
+			function ( array $body, ProductInput $input ): ProductInput {
+				return $input;
 			}
-		};
+		);
+	}
+
+	/**
+	 * Run product sub-requests in concurrent multipart batches, demuxing each
+	 * sub-response back to a per-input success (ProductInput) or failure
+	 * (MerchantApiException). A whole-batch failure marks every input in it as failed.
+	 *
+	 * @param ProductInput[] $inputs
+	 * @param int            $concurrency   Concurrent batch requests.
+	 * @param string         $operation     Method label used on sub-request failures.
+	 * @param callable       $build_request fn(int $index, ProductInput $input): array{method: string, path: string, body?: array}
+	 * @param callable       $on_success    fn(array $body, ProductInput $input): ProductInput
+	 *
+	 * @return array{successes: array<int, ProductInput>, failures: array<int, MerchantApiException>}
+	 */
+	private function run_in_batches( array $inputs, int $concurrency, string $operation, callable $build_request, callable $on_success ): array {
+		$client        = $this->client;
+		$index_batches = array_chunk( array_keys( $inputs ), $this->get_batch_size() );
 
 		$successes = [];
 		$failures  = [];
 
+		$promises = function () use ( $index_batches, $inputs, $client, $build_request ) {
+			foreach ( $index_batches as $batch_num => $indices ) {
+				$requests = [];
+				foreach ( $indices as $index ) {
+					$requests[ $index ] = $build_request( $index, $inputs[ $index ] );
+				}
+
+				yield $batch_num => $client->batch_async( $requests );
+			}
+		};
+
 		( new EachPromise(
-			$requests(),
+			$promises(),
 			[
 				'concurrency' => $concurrency,
-				'fulfilled'   => function ( array $body, int $index ) use ( &$successes, $inputs ) {
-					$successes[ $index ] = $inputs[ $index ];
+				'fulfilled'   => function ( array $batch_results, $batch_num ) use ( &$successes, &$failures, $inputs, $index_batches, $on_success, $operation ) {
+					foreach ( $batch_results as $index => $result ) {
+						// Ignore a sub-response for an id that was not in this batch (guards
+						// against a malformed or duplicate Content-ID in the response).
+						if ( ! isset( $inputs[ $index ] ) ) {
+							continue;
+						}
+
+						if ( $result['status'] >= 200 && $result['status'] < 300 ) {
+							$successes[ $index ] = $on_success( $result['body'], $inputs[ $index ] );
+						} else {
+							$failures[ $index ] = new MerchantApiException( $result['status'], $result['body'], $operation );
+						}
+					}
+
+					// A requested sub-request missing from the parsed response is a retryable
+					// failure, never a silent no-op.
+					foreach ( array_diff( $index_batches[ $batch_num ], array_keys( $batch_results ) ) as $index ) {
+						$failures[ $index ] = new MerchantApiException(
+							500,
+							[ 'error' => [ 'message' => 'No sub-response returned for this product in the batch response.' ] ],
+							$operation
+						);
+					}
 				},
-				'rejected'    => function ( $reason, int $index ) use ( &$failures ) {
-					$failures[ $index ] = $reason;
+				'rejected'    => function ( $reason, $batch_num ) use ( &$failures, $index_batches ) {
+					foreach ( $index_batches[ $batch_num ] as $index ) {
+						$failures[ $index ] = $reason;
+					}
 				},
 			]
 		) )->promise()->wait();
@@ -275,6 +322,15 @@ public function delete_many( array $inputs, int $concurrency = 10 ): array {
 		];
 	}
 
+	/**
+	 * Number of product sub-requests to pack into a single batch HTTP request.
+	 *
+	 * @return int
+	 */
+	protected function get_batch_size(): int {
+		return max( 1, (int) apply_filters( 'woocommerce_gla_mapi_batch_size', 50 ) );
+	}
+
 	/**
 	 * Build the productInputs.insert path with the resolved data source.
 	 *