M7 — Delivery, scale & more FeedTypes (final milestone)

README.md

@@ -93,11 +93,28 @@ one or more channels.
 After that go to your console and run this command:
 
 ```bash
-$ php bin/console setono:sylius-feed:process
+$ php bin/console setono:feed:process --all
 ```
 
+`setono:feed:process` dispatches feed generation. Without options (or with `--all`) it processes
+every **enabled** feed; pass `--feed=CODE` to process a single feed. Add `--preview[=N]` for a
+dry-run that prints the include funnel and a few sample rows without writing a file, and `--audit`
+to also print per-attribute fill rates and soft warnings.
+
 If you haven't changed any configuration, there should be a feed with your products inside the `/var/storage/setono_sylius_feed/feed` directory.
 
+### Scheduling (cron)
+
+There is no bundled scheduler — run the process command from your own crontab. For example, to
+regenerate every enabled feed hourly:
+
+```cron
+0 * * * * cd /path/to/your/project && php bin/console setono:feed:process --all >> var/log/feed.log 2>&1
+```
+
+When the plugin's messages are routed to an async transport (recommended, see Step 6), this command
+only *dispatches* the work; make sure a worker is consuming the transport (e.g. `messenger:consume`).
+
 
 [ico-version]: https://poser.pugx.org/setono/sylius-feed-plugin/v/stable
 [ico-license]: https://poser.pugx.org/setono/sylius-feed-plugin/license

composer.json

@@ -13,8 +13,9 @@
         "doctrine/persistence": "^1.3 || ^2.0 || ^3.0",
         "knplabs/knp-menu": "^3.1",
         "league/csv": "^9.6",
-        "league/flysystem": "^2.1 || ^3.0",
+        "league/flysystem": "^3.15",
         "league/flysystem-bundle": "^2.4 || ^3.0",
+        "league/flysystem-local": "^3.15",
         "liip/imagine-bundle": "^2.4",
         "psr/event-dispatcher": "^1.0",
         "psr/log": "^1.0 || ^2.0 || ^3.0",
@@ -56,16 +57,23 @@
         "doctrine/doctrine-bundle": "^2.11",
         "jms/serializer-bundle": "^4.2",
         "lexik/jwt-authentication-bundle": "^2.17",
+        "nikic/php-parser": "^5.0",
         "setono/sylius-plugin-pack": "~1.14.2",
         "sylius-labs/polyfill-symfony-security": "^1.1.2",
         "symfony/debug-bundle": "^6.4",
         "symfony/dotenv": "^6.4",
         "symfony/intl": "^6.4",
         "symfony/property-info": "^6.4",
+        "symfony/translation": "^6.4.3",
         "symfony/web-profiler-bundle": "^6.4",
         "symfony/webpack-encore-bundle": "^1.17.2",
         "willdurand/negotiation": "^3.1"
     },
+    "suggest": {
+        "league/flysystem-aws-s3-v3": "To deliver generated feeds to Amazon S3 (the 's3' delivery transport)",
+        "league/flysystem-ftp": "To deliver generated feeds over FTP (the 'ftp' delivery transport)",
+        "league/flysystem-sftp-v3": "To deliver generated feeds over SFTP (the 'sftp' delivery transport)"
+    },
     "prefer-stable": true,
     "autoload": {
         "psr-4": {

src/Command/ProcessFeedCommand.php

@@ -43,6 +43,7 @@ protected function configure(): void
     {
         $this
             ->addOption('feed', null, InputOption::VALUE_REQUIRED, 'Process only the feed with this code')
+            ->addOption('all', null, InputOption::VALUE_NONE, 'Process every enabled feed (the default when --feed is omitted); the explicit form to use from cron')
             ->addOption('preview', null, InputOption::VALUE_OPTIONAL, 'Dry-run: print the funnel + a few sample rows instead of generating (optionally set the sample size, default ' . self::DEFAULT_PREVIEW_LIMIT . ')', false)
             ->addOption('audit', null, InputOption::VALUE_NONE, 'Dry-run: also print fill rates and soft warnings (implies --preview)')
         ;

src/Context/MessageContextFactory.php

@@ -0,0 +1,27 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Setono\SyliusFeedPlugin\Context;
+
+use Sylius\Component\Core\Model\ChannelInterface;
+use Sylius\Component\Resource\Repository\RepositoryInterface;
+
+final class MessageContextFactory implements MessageContextFactoryInterface
+{
+    public function __construct(
+        private readonly RepositoryInterface $channelRepository,
+    ) {
+    }
+
+    public function create(?string $channelCode, ?string $locale, ?string $currencyCode): FeedContext
+    {
+        $channel = null;
+        if (null !== $channelCode) {
+            $candidate = $this->channelRepository->findOneBy(['code' => $channelCode]);
+            $channel = $candidate instanceof ChannelInterface ? $candidate : null;
+        }
+
+        return new FeedContext($channel, $locale, $currencyCode);
+    }
+}

src/Context/MessageContextFactoryInterface.php

@@ -0,0 +1,14 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Setono\SyliusFeedPlugin\Context;
+
+/**
+ * Rebuilds a {@see FeedContext} from the transport-safe dimension codes carried on a generation
+ * message (§6.2). Shared by every generation handler so a context is reconstructed identically.
+ */
+interface MessageContextFactoryInterface
+{
+    public function create(?string $channelCode, ?string $locale, ?string $currencyCode): FeedContext;
+}

src/DataSource/CustomerDataSource.php

@@ -0,0 +1,82 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Setono\SyliusFeedPlugin\DataSource;
+
+use Doctrine\ORM\QueryBuilder;
+use Doctrine\Persistence\ManagerRegistry;
+use Setono\Doctrine\ORMTrait;
+use Setono\SyliusFeedPlugin\Context\FeedContext;
+use Setono\SyliusFeedPlugin\Doctrine\BatchIterator;
+use Setono\SyliusFeedPlugin\Filter\FilterSet;
+use Setono\SyliusFeedPlugin\Generator\ChunkRange;
+
+/**
+ * Streams every customer — one row per customer (§8.3). Unlike a product, a customer has no
+ * enabled/state concept to guard on, and the feed type declares no scope dimensions, so this data
+ * source needs neither a query-level filter nor a context-dependent constraint: it is the
+ * thinnest possible data source, proving the engine does not require either (§6.3, §8.3).
+ * Iterates in bounded-memory batches via {@see BatchIterator} (clears the entity manager every
+ * batch).
+ */
+final class CustomerDataSource implements DataSourceInterface
+{
+    use ORMTrait;
+    use IdRangePartitionTrait;
+
+    private const BATCH_SIZE = 1000;
+
+    /**
+     * @param class-string $resourceClass
+     */
+    public function __construct(
+        ManagerRegistry $managerRegistry,
+        private readonly string $resourceClass,
+    ) {
+        $this->managerRegistry = $managerRegistry;
+    }
+
+    public function getResourceClass(): string
+    {
+        return $this->resourceClass;
+    }
+
+    public function getItems(FeedContext $context, FilterSet $filters): iterable
+    {
+        return BatchIterator::iterate(
+            $this->createQueryBuilder()->getQuery(),
+            $this->getManager($this->resourceClass),
+            self::BATCH_SIZE,
+        );
+    }
+
+    public function count(FeedContext $context, FilterSet $filters): int
+    {
+        return (int) $this->createQueryBuilder()
+            ->select('COUNT(c.id)')
+            ->getQuery()
+            ->getSingleScalarResult();
+    }
+
+    public function getIdRange(FeedContext $context, FilterSet $filters): ?ChunkRange
+    {
+        return $this->resolveIdRange($this->createQueryBuilder(), 'c');
+    }
+
+    public function getItemsInRange(FeedContext $context, FilterSet $filters, ChunkRange $range): iterable
+    {
+        return BatchIterator::iterate(
+            $this->constrainToRange($this->createQueryBuilder(), 'c', $range)->getQuery(),
+            $this->getManager($this->resourceClass),
+            self::BATCH_SIZE,
+        );
+    }
+
+    private function createQueryBuilder(): QueryBuilder
+    {
+        return $this->getManager($this->resourceClass)
+            ->getRepository($this->resourceClass)
+            ->createQueryBuilder('c');
+    }
+}

src/DataSource/DataSourceInterface.php

@@ -6,6 +6,7 @@
 
 use Setono\SyliusFeedPlugin\Context\FeedContext;
 use Setono\SyliusFeedPlugin\Filter\FilterSet;
+use Setono\SyliusFeedPlugin\Generator\ChunkRange;
 
 /**
  * Yields the source entities for a feed type, in batches, given a context + filters (§5).
@@ -25,4 +26,21 @@ public function getResourceClass(): string;
     public function getItems(FeedContext $context, FilterSet $filters): iterable;
 
     public function count(FeedContext $context, FilterSet $filters): int;
+
+    /**
+     * The inclusive `[min, max]` id bounds of the entities this source would yield for the given
+     * context and filters, or null when the source cannot be partitioned by id (e.g. an in-memory or
+     * collection source) or would yield nothing. Used to plan chunk ranges for fan-out generation of
+     * a large source (§6.3); returning null keeps that source on the inline path.
+     */
+    public function getIdRange(FeedContext $context, FilterSet $filters): ?ChunkRange;
+
+    /**
+     * Streams only the entities whose id falls within the given chunk range, in ascending id order,
+     * so a fan-out chunk yields exactly its slice of the source (§6.3). MUST stream/batch. The
+     * ordered concatenation of every range must reproduce the {@see getItems()} order exactly.
+     *
+     * @return iterable<object>
+     */
+    public function getItemsInRange(FeedContext $context, FilterSet $filters, ChunkRange $range): iterable;
 }

src/DataSource/IdRangePartitionTrait.php

@@ -0,0 +1,53 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Setono\SyliusFeedPlugin\DataSource;
+
+use Doctrine\ORM\QueryBuilder;
+use Setono\SyliusFeedPlugin\Generator\ChunkRange;
+
+/**
+ * Shared id-range partitioning for the Doctrine-backed data sources (§6.3): resolves the `[min, max]`
+ * id bounds of a source's filtered query and constrains a query to a single chunk range, ordered by
+ * id, so a fan-out chunk yields exactly its slice and its ordered concatenation matches the inline
+ * stream byte-for-byte.
+ */
+trait IdRangePartitionTrait
+{
+    /**
+     * The inclusive `[min, max]` id bounds of the given query, or null when it would yield nothing.
+     */
+    private function resolveIdRange(QueryBuilder $queryBuilder, string $alias): ?ChunkRange
+    {
+        $row = (clone $queryBuilder)
+            ->select(sprintf('MIN(%1$s.id) AS min_id, MAX(%1$s.id) AS max_id', $alias))
+            ->getQuery()
+            ->getSingleResult();
+
+        if (!is_array($row)) {
+            return null;
+        }
+
+        $min = $row['min_id'] ?? null;
+        $max = $row['max_id'] ?? null;
+
+        if (!is_numeric($min) || !is_numeric($max)) {
+            return null;
+        }
+
+        return new ChunkRange((int) $min, (int) $max);
+    }
+
+    /**
+     * Constrains the query to a single chunk range, ordered by id ascending.
+     */
+    private function constrainToRange(QueryBuilder $queryBuilder, string $alias, ChunkRange $range): QueryBuilder
+    {
+        return $queryBuilder
+            ->andWhere(sprintf('%s.id BETWEEN :chunkStart AND :chunkEnd', $alias))
+            ->addOrderBy(sprintf('%s.id', $alias), 'ASC')
+            ->setParameter('chunkStart', $range->start)
+            ->setParameter('chunkEnd', $range->end);
+    }
+}

src/DataSource/OrderDataSource.php

@@ -0,0 +1,94 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Setono\SyliusFeedPlugin\DataSource;
+
+use Doctrine\ORM\QueryBuilder;
+use Doctrine\Persistence\ManagerRegistry;
+use Setono\Doctrine\ORMTrait;
+use Setono\SyliusFeedPlugin\Context\FeedContext;
+use Setono\SyliusFeedPlugin\Doctrine\BatchIterator;
+use Setono\SyliusFeedPlugin\Filter\FilterSet;
+use Setono\SyliusFeedPlugin\Generator\ChunkRange;
+use Sylius\Component\Core\Model\ChannelInterface;
+use Sylius\Component\Order\Model\OrderInterface;
+
+/**
+ * Streams completed orders — one row per order (§8.2). "Completed" excludes carts (state !=
+ * `cart`); an order in any other state (new, cancelled, fulfilled, …) is included, since a CSV
+ * export is typically interested in the full order history, not just fulfilled ones. Iterates in
+ * bounded-memory batches via {@see BatchIterator} (clears the entity manager every batch), proving
+ * the same engine that streams a product catalog streams an order export just as well (§6.3).
+ */
+final class OrderDataSource implements DataSourceInterface
+{
+    use ORMTrait;
+    use IdRangePartitionTrait;
+
+    private const BATCH_SIZE = 1000;
+
+    /**
+     * @param class-string $resourceClass
+     */
+    public function __construct(
+        ManagerRegistry $managerRegistry,
+        private readonly string $resourceClass,
+    ) {
+        $this->managerRegistry = $managerRegistry;
+    }
+
+    public function getResourceClass(): string
+    {
+        return $this->resourceClass;
+    }
+
+    public function getItems(FeedContext $context, FilterSet $filters): iterable
+    {
+        return BatchIterator::iterate(
+            $this->createQueryBuilder($context)->getQuery(),
+            $this->getManager($this->resourceClass),
+            self::BATCH_SIZE,
+        );
+    }
+
+    public function count(FeedContext $context, FilterSet $filters): int
+    {
+        return (int) $this->createQueryBuilder($context)
+            ->select('COUNT(o.id)')
+            ->getQuery()
+            ->getSingleScalarResult();
+    }
+
+    public function getIdRange(FeedContext $context, FilterSet $filters): ?ChunkRange
+    {
+        return $this->resolveIdRange($this->createQueryBuilder($context), 'o');
+    }
+
+    public function getItemsInRange(FeedContext $context, FilterSet $filters, ChunkRange $range): iterable
+    {
+        return BatchIterator::iterate(
+            $this->constrainToRange($this->createQueryBuilder($context), 'o', $range)->getQuery(),
+            $this->getManager($this->resourceClass),
+            self::BATCH_SIZE,
+        );
+    }
+
+    private function createQueryBuilder(FeedContext $context): QueryBuilder
+    {
+        $queryBuilder = $this->getManager($this->resourceClass)
+            ->getRepository($this->resourceClass)
+            ->createQueryBuilder('o')
+            ->andWhere('o.state != :cartState')
+            ->setParameter('cartState', OrderInterface::STATE_CART);
+
+        $channel = $context->getChannel();
+        if ($channel instanceof ChannelInterface) {
+            $queryBuilder
+                ->andWhere('o.channel = :channel')
+                ->setParameter('channel', $channel);
+        }
+
+        return $queryBuilder;
+    }
+}