Feature/add cursor for long request

.github/workflows/quality.yaml

@@ -12,7 +12,7 @@ jobs:
                 run: |
                     wget -q https://cs.symfony.com/download/php-cs-fixer-v3.phar -O php-cs-fixer
                     chmod a+x php-cs-fixer
-                    PHP_CS_FIXER_IGNORE_ENV=true ./php-cs-fixer fix src --dry-run
+                    ./php-cs-fixer fix src --dry-run
 
     phpstan:
         runs-on: ubuntu-latest

phpstan.neon

@@ -0,0 +1,3 @@
+parameters:
+    level: 5
+    treatPhpDocTypesAsCertain: false

src/CategoryLookup.php

@@ -5,6 +5,7 @@
 namespace Kiboko\Component\Flow\Magento2;
 
 use Kiboko\Component\Bucket\AcceptanceResultBucket;
+use Kiboko\Component\Bucket\EmptyResultBucket;
 use Kiboko\Component\Bucket\RejectionResultBucket;
 use Kiboko\Contract\Mapping\CompiledMapperInterface;
 use Kiboko\Contract\Pipeline\TransformerInterface;
@@ -24,7 +25,7 @@ public function __construct(
 
     public function transform(): \Generator
     {
-        $line = yield;
+        $line = yield new EmptyResultBucket();
         while (true) {
             if (null === $line[$this->mappingField]) {
                 $line = yield new AcceptanceResultBucket($line);

src/CustomerExtractor.php

@@ -6,64 +6,91 @@
 
 use Kiboko\Component\Bucket\AcceptanceResultBucket;
 use Kiboko\Component\Bucket\RejectionResultBucket;
+use Kiboko\Component\Flow\Magento2\Filter\FilterInterface;
 use Kiboko\Contract\Bucket\ResultBucketInterface;
 use Kiboko\Contract\Pipeline\ExtractorInterface;
+use Psr\Http\Client\NetworkExceptionInterface;
 
-final class CustomerExtractor implements ExtractorInterface
+final readonly class CustomerExtractor implements ExtractorInterface
 {
-    private array $queryParameters = [
-        'searchCriteria[currentPage]' => 1,
-        'searchCriteria[pageSize]' => 100,
-    ];
-
     public function __construct(
-        private readonly \Psr\Log\LoggerInterface $logger,
-        private readonly \Kiboko\Magento\V2_1\Client|\Kiboko\Magento\V2_2\Client|\Kiboko\Magento\V2_3\Client|\Kiboko\Magento\V2_4\Client $client,
-        private readonly int $pageSize = 100,
-        /** @var FilterGroup[] $filters */
-        private readonly array $filters = [],
+        private \Psr\Log\LoggerInterface $logger,
+        private \Kiboko\Magento\V2_1\Client|\Kiboko\Magento\V2_2\Client|\Kiboko\Magento\V2_3\Client|\Kiboko\Magento\V2_4\Client $client,
+        private QueryParameters $queryParameters,
+        private int $pageSize = 100,
     ) {
     }
 
-    private function compileQueryParameters(int $currentPage = 1): array
+    private function walkFilterVariants(int $currentPage = 1): \Traversable
     {
-        $parameters = $this->queryParameters;
-        $parameters['searchCriteria[currentPage]'] = $currentPage;
-        $parameters['searchCriteria[pageSize]'] = $this->pageSize;
-
-        $filters = array_map(fn (FilterGroup $item, int $key) => $item->compileFilters($key), $this->filters, array_keys($this->filters));
+        yield from [
+            ...$this->queryParameters->walkVariants([]),
+            ...[
+                'searchCriteria[currentPage]' => $currentPage,
+                'searchCriteria[pageSize]' => $this->pageSize,
+            ],
+        ];
+    }
 
-        return array_merge($parameters, ...$filters);
+    private function applyPagination(array $parameters, int $currentPage, int $pageSize): array
+    {
+        return [
+            ...$parameters,
+            ...[
+                'searchCriteria[currentPage]' => $currentPage,
+                'searchCriteria[pageSize]' => $pageSize,
+            ],
+        ];
     }
 
     public function extract(): iterable
     {
+        $currentPage = null;
+        $pageCount = null;
         try {
-            $response = $this->client->customerCustomerRepositoryV1GetListGet(
-                queryParameters: $this->compileQueryParameters(),
-            );
-
-            if (!$response instanceof \Kiboko\Magento\V2_1\Model\CustomerDataCustomerSearchResultsInterface
-                && !$response instanceof \Kiboko\Magento\V2_2\Model\CustomerDataCustomerSearchResultsInterface
-                && !$response instanceof \Kiboko\Magento\V2_3\Model\CustomerDataCustomerSearchResultsInterface
-                && !$response instanceof \Kiboko\Magento\V2_4\Model\CustomerDataCustomerSearchResultsInterface
-            ) {
-                return;
-            }
-
-            yield $this->processResponse($response);
-
-            $currentPage = 1;
-            $pageCount = ceil($response->getTotalCount() / $this->pageSize);
-            while ($currentPage++ < $pageCount) {
+            foreach ($this->queryParameters->walkVariants([]) as $parameters) {
+                $currentPage = 1;
                 $response = $this->client->customerCustomerRepositoryV1GetListGet(
-                    queryParameters: $this->compileQueryParameters($currentPage),
+                    queryParameters: $this->applyPagination(iterator_to_array($parameters), $currentPage, $this->pageSize),
                 );
+                $pageCount = (int) ceil($response->getTotalCount() / $this->pageSize);
+
+                if (!$response instanceof \Kiboko\Magento\V2_1\Model\CustomerDataCustomerSearchResultsInterface
+                    && !$response instanceof \Kiboko\Magento\V2_2\Model\CustomerDataCustomerSearchResultsInterface
+                    && !$response instanceof \Kiboko\Magento\V2_3\Model\CustomerDataCustomerSearchResultsInterface
+                    && !$response instanceof \Kiboko\Magento\V2_4\Model\CustomerDataCustomerSearchResultsInterface
+                ) {
+                    return;
+                }
 
                 yield $this->processResponse($response);
+
+                while ($currentPage++ < $pageCount) {
+                    $response = $this->client->customerCustomerRepositoryV1GetListGet(
+                        queryParameters: $this->applyPagination(iterator_to_array($parameters), $currentPage, $this->pageSize),
+                    );
+
+                    yield $this->processResponse($response);
+                }
             }
+        } catch (NetworkExceptionInterface $exception) {
+            $this->logger->alert(
+                $exception->getMessage(),
+                [
+                    'exception' => $exception,
+                    'context' => [
+                        'path' => 'customer',
+                        'method' => 'get',
+                        'queryParameters' => $this->walkFilterVariants(),
+                    ],
+                ],
+            );
+            yield new RejectionResultBucket(
+                'There are some network difficulties. We could not properly connect to the Magento API. There is nothing we could no to fix this currently. Please contact the Magento administrator.',
+                $exception,
+            );
         } catch (\Exception $exception) {
-            $this->logger->alert($exception->getMessage(), ['exception' => $exception]);
+            $this->logger->critical($exception->getMessage(), ['exception' => $exception]);
         }
     }
 
@@ -74,7 +101,7 @@ private function processResponse($response): ResultBucketInterface
             || $response instanceof \Kiboko\Magento\V2_3\Model\ErrorResponse
             || $response instanceof \Kiboko\Magento\V2_4\Model\ErrorResponse
         ) {
-            return new RejectionResultBucket($response);
+            return new RejectionResultBucket($response->getMessage(), null, $response);
         }
 
         return new AcceptanceResultBucket(...$response->getItems());

src/Filter/ArrayFilter.php

@@ -0,0 +1,30 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Kiboko\Component\Flow\Magento2\Filter;
+
+final class ArrayFilter implements FilterInterface, \IteratorAggregate
+{
+    public function __construct(
+        public string $field,
+        public string $conditionType,
+        public array $value,
+        private readonly int $threshold = 200
+    ) {}
+
+    /**
+     * @return \Traversable<int, {field: string, value: string, conditionType: string}>
+     */
+    public function getIterator(): \Traversable
+    {
+        $length = count($this->value);
+        for ($offset = 0; $offset < $length; $offset += $this->threshold) {
+            yield [
+                'field' => $this->field,
+                'value' => implode(',', array_slice($this->value, $offset, $this->threshold, false)),
+                'conditionType' => $this->conditionType,
+            ];
+        }
+    }
+}

src/Filter/FilterInterface.php

@@ -0,0 +1,12 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Kiboko\Component\Flow\Magento2\Filter;
+
+/**
+ * @extends \Traversable<int, {field: string, value: string, conditionType: string}>
+ */
+interface FilterInterface extends \Traversable
+{
+}

src/Filter/ScalarFilter.php

@@ -0,0 +1,23 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Kiboko\Component\Flow\Magento2\Filter;
+
+final class ScalarFilter implements FilterInterface, \IteratorAggregate
+{
+    public function __construct(
+        public string $field,
+        public string $conditionType,
+        public bool|int|float|string|\DateTimeInterface $value,
+    ) {}
+
+    public function getIterator(): \Traversable
+    {
+        yield [
+            'field' => $this->field,
+            'value' => $this->value,
+            'conditionType' => $this->conditionType,
+        ];
+    }
+}

src/FilterGroup.php

@@ -4,48 +4,78 @@
 
 namespace Kiboko\Component\Flow\Magento2;
 
+use Kiboko\Component\Flow\Magento2\Filter\FilterInterface;
+use Kiboko\Component\Flow\Magento2\Filter\ScalarFilter;
+
 class FilterGroup
 {
     private array $filters = [];
 
-    public function withFilter(Filter $filter): self
+    public function withFilter(FilterInterface $filter): self
     {
-        $this->filters[] = [
-            'field' => $filter->field,
-            'value' => $filter->value,
-            'condition_type' => $filter->conditionType,
-        ];
+        $this->filters[] = $filter;
 
         return $this;
     }
 
-    public function withFilters(Filter ...$filters): self
+    public function withFilters(FilterInterface ...$filter): self
     {
-        array_walk($filters, fn (Filter $filter) => $this->filters[] = [
-            'field' => $filter->field,
-            'value' => $filter->value,
-            'condition_type' => $filter->conditionType,
-        ]);
+        foreach ($filter as $item) {
+            $this->filters[] = $item;
+        }
 
         return $this;
     }
 
-    public function compileFilters(int $groupIndex = 0): array
+    /**
+     * @return \Traversable<int, array>
+     */
+    public function walkFilters(array $parameters, int $groupIndex = 0): \Traversable
+    {
+        if (count($this->filters) < 1) {
+            return;
+        }
+
+        yield from $this->buildFilters($parameters, $groupIndex, 1, ...$this->filters);
+    }
+
+    private function buildFilters(array $parameters, int $groupIndex, int $filterIndex, FilterInterface $first, FilterInterface ...$next): \Traversable
+    {
+        foreach ($first as $current) {
+            $childParameters = [
+                ...$parameters,
+                ...[
+                    sprintf('searchCriteria[filterGroups][%s][filters][%s][field]', $groupIndex, $filterIndex) => $current['field'],
+                    sprintf('searchCriteria[filterGroups][%s][filters][%s][value]', $groupIndex, $filterIndex) => $current['value'],
+                    sprintf('searchCriteria[filterGroups][%s][filters][%s][conditionType]', $groupIndex, $filterIndex) => $current['conditionType'],
+                ]
+            ];
+
+            if (count($next) >= 1) {
+                yield from $this->buildFilters($childParameters, $groupIndex, $filterIndex + 1, ...$next);
+            } else {
+                yield $childParameters;
+            }
+        }
+    }
+
+    public function greaterThan(string $field, int|float|string|\DateTimeInterface $value): self
+    {
+        return $this->withFilter(new ScalarFilter($field, 'gt', $value));
+    }
+
+    public function lowerThan(string $field, int|float|string|\DateTimeInterface $value): self
     {
-        return array_merge(...array_map(fn (array $item, int $key) => [
-            sprintf('searchCriteria[filterGroups][%s][filters][%s][field]', $groupIndex, $key) => $item['field'],
-            sprintf('searchCriteria[filterGroups][%s][filters][%s][value]', $groupIndex, $key) => $item['value'],
-            sprintf('searchCriteria[filterGroups][%s][filters][%s][conditionType]', $groupIndex, $key) => $item['condition_type'],
-        ], $this->filters, array_keys($this->filters)));
+        return $this->withFilter(new ScalarFilter($field, 'lt', $value));
     }
 
-    public function greaterThan(string $field, mixed $value): self
+    public function greaterThanOrEqual(string $field, int|float|string|\DateTimeInterface $value): self
     {
-        return $this->withFilter(new Filter($field, 'gt', $value));
+        return $this->withFilter(new ScalarFilter($field, 'gteq', $value));
     }
 
-    public function greaterThanEqual(string $field, mixed $value): self
+    public function lowerThanOrEqual(string $field, int|float|string|\DateTimeInterface $value): self
     {
-        return $this->withFilter(new Filter($field, 'gteq', $value));
+        return $this->withFilter(new ScalarFilter($field, 'lteq', $value));
     }
 }