FEATURE: Subscription Engine

Neos.ContentGraph.DoctrineDbalAdapter/src/DoctrineDbalContentGraphProjectionFactory.php

@@ -8,7 +8,7 @@
 use Neos\ContentGraph\DoctrineDbalAdapter\Domain\Repository\DimensionSpacePointsRepository;
 use Neos\ContentGraph\DoctrineDbalAdapter\Domain\Repository\NodeFactory;
 use Neos\ContentGraph\DoctrineDbalAdapter\Domain\Repository\ProjectionContentGraph;
-use Neos\ContentRepository\Core\Factory\ProjectionFactoryDependencies;
+use Neos\ContentRepository\Core\Factory\SubscriberFactoryDependencies;
 use Neos\ContentRepository\Core\Projection\ContentGraph\ContentGraphProjectionFactoryInterface;
 
 /**
@@ -24,8 +24,7 @@ public function __construct(
     }
 
     public function build(
-        ProjectionFactoryDependencies $projectionFactoryDependencies,
-        array $options,
+        SubscriberFactoryDependencies $projectionFactoryDependencies,
     ): DoctrineDbalContentGraphProjection {
         $tableNames = ContentGraphTableNames::create(
             $projectionFactoryDependencies->contentRepositoryId

Neos.ContentGraph.PostgreSQLAdapter/src/HypergraphProjectionFactory.php

@@ -7,7 +7,7 @@
 use Doctrine\DBAL\Connection;
 use Neos\ContentGraph\PostgreSQLAdapter\Domain\Projection\HypergraphProjection;
 use Neos\ContentGraph\PostgreSQLAdapter\Domain\Repository\NodeFactory;
-use Neos\ContentRepository\Core\Factory\ProjectionFactoryDependencies;
+use Neos\ContentRepository\Core\Factory\SubscriberFactoryDependencies;
 use Neos\ContentRepository\Core\Projection\ContentGraph\ContentGraphProjectionFactoryInterface;
 use Neos\ContentRepository\Core\SharedModel\ContentRepository\ContentRepositoryId;
 
@@ -28,8 +28,7 @@ public static function graphProjectionTableNamePrefix(
     }
 
     public function build(
-        ProjectionFactoryDependencies $projectionFactoryDependencies,
-        array $options,
+        SubscriberFactoryDependencies $projectionFactoryDependencies,
     ): HypergraphProjection {
         $tableNamePrefix = self::graphProjectionTableNamePrefix(
             $projectionFactoryDependencies->contentRepositoryId

Neos.ContentRepository.BehavioralTests/Classes/ProjectionRaceConditionTester/RaceTrackerCatchUpHook.php

@@ -17,7 +17,8 @@
 use Neos\ContentRepository\BehavioralTests\ProjectionRaceConditionTester\Dto\TraceEntries;
 use Neos\ContentRepository\BehavioralTests\ProjectionRaceConditionTester\Dto\TraceEntryType;
 use Neos\ContentRepository\Core\EventStore\EventInterface;
-use Neos\ContentRepository\Core\Projection\CatchUpHookInterface;
+use Neos\ContentRepository\Core\Projection\CatchUpHook\CatchUpHookInterface;
+use Neos\ContentRepository\Core\Subscription\SubscriptionStatus;
 use Neos\EventStore\Model\EventEnvelope;
 use Neos\Flow\Annotations as Flow;
 
@@ -107,7 +108,7 @@ final class RaceTrackerCatchUpHook implements CatchUpHookInterface
     protected $configuration;
     private bool $inCriticalSection = false;
 
-    public function onBeforeCatchUp(): void
+    public function onBeforeCatchUp(SubscriptionStatus $subscriptionStatus): void
     {
         RedisInterleavingLogger::connect($this->configuration['redis']['host'], $this->configuration['redis']['port']);
     }
@@ -126,16 +127,12 @@ public function onAfterEvent(EventInterface $eventInstance, EventEnvelope $event
     {
     }
 
-    public function onBeforeBatchCompleted(): void
+    public function onAfterCatchUp(): void
     {
         // we only want to track relevant lock release calls (i.e. if we were in the event processing loop before)
         if ($this->inCriticalSection) {
             $this->inCriticalSection = false;
             RedisInterleavingLogger::trace(TraceEntryType::LockWillBeReleasedIfItWasAcquiredBefore);
         }
     }
-
-    public function onAfterCatchUp(): void
-    {
-    }
 }

Neos.ContentRepository.BehavioralTests/Classes/ProjectionRaceConditionTester/RaceTrackerCatchUpHookFactory.php

@@ -14,9 +14,9 @@
 
 namespace Neos\ContentRepository\BehavioralTests\ProjectionRaceConditionTester;
 
-use Neos\ContentRepository\Core\Projection\CatchUpHookFactoryDependencies;
-use Neos\ContentRepository\Core\Projection\CatchUpHookFactoryInterface;
-use Neos\ContentRepository\Core\Projection\CatchUpHookInterface;
+use Neos\ContentRepository\Core\Projection\CatchUpHook\CatchUpHookFactoryDependencies;
+use Neos\ContentRepository\Core\Projection\CatchUpHook\CatchUpHookFactoryInterface;
+use Neos\ContentRepository\Core\Projection\CatchUpHook\CatchUpHookInterface;
 use Neos\ContentRepository\Core\Projection\ContentGraph\ContentGraphReadModelInterface;
 
 /**

Neos.ContentRepository.Core/Classes/ContentRepository.php

@@ -18,25 +18,16 @@
 use Neos\ContentRepository\Core\CommandHandler\CommandInterface;
 use Neos\ContentRepository\Core\Dimension\ContentDimensionSourceInterface;
 use Neos\ContentRepository\Core\DimensionSpace\InterDimensionalVariationGraph;
-use Neos\ContentRepository\Core\EventStore\EventNormalizer;
 use Neos\ContentRepository\Core\EventStore\EventPersister;
 use Neos\ContentRepository\Core\EventStore\EventsToPublish;
 use Neos\ContentRepository\Core\EventStore\InitiatingEventMetadata;
 use Neos\ContentRepository\Core\Factory\ContentRepositoryFactory;
 use Neos\ContentRepository\Core\NodeType\NodeTypeManager;
-use Neos\ContentRepository\Core\Projection\CatchUp;
-use Neos\ContentRepository\Core\Projection\CatchUpHookFactoryDependencies;
-use Neos\ContentRepository\Core\Projection\CatchUpOptions;
 use Neos\ContentRepository\Core\Projection\ContentGraph\ContentGraphInterface;
-use Neos\ContentRepository\Core\Projection\ContentGraph\ContentGraphProjectionInterface;
 use Neos\ContentRepository\Core\Projection\ContentGraph\ContentGraphReadModelInterface;
-use Neos\ContentRepository\Core\Projection\ProjectionInterface;
-use Neos\ContentRepository\Core\Projection\ProjectionsAndCatchUpHooks;
 use Neos\ContentRepository\Core\Projection\ProjectionStateInterface;
-use Neos\ContentRepository\Core\Projection\ProjectionStatuses;
-use Neos\ContentRepository\Core\Projection\WithMarkStaleInterface;
+use Neos\ContentRepository\Core\Projection\ProjectionStates;
 use Neos\ContentRepository\Core\SharedModel\ContentRepository\ContentRepositoryId;
-use Neos\ContentRepository\Core\SharedModel\ContentRepository\ContentRepositoryStatus;
 use Neos\ContentRepository\Core\SharedModel\Exception\WorkspaceDoesNotExist;
 use Neos\ContentRepository\Core\SharedModel\User\UserIdProviderInterface;
 use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStream;
@@ -45,45 +36,34 @@
 use Neos\ContentRepository\Core\SharedModel\Workspace\Workspace;
 use Neos\ContentRepository\Core\SharedModel\Workspace\WorkspaceName;
 use Neos\ContentRepository\Core\SharedModel\Workspace\Workspaces;
-use Neos\EventStore\EventStoreInterface;
-use Neos\EventStore\Model\EventEnvelope;
-use Neos\EventStore\Model\EventStream\VirtualStreamName;
 use Psr\Clock\ClockInterface;
 
 /**
  * Main Entry Point to the system. Encapsulates the full event-sourced Content Repository.
  *
  * Use this to:
- * - set up the necessary database tables and contents via {@see ContentRepository::setUp()}
- * - send commands to the system (to mutate state) via {@see ContentRepository::handle()}
- * - access projection state (to read state) via {@see ContentRepository::projectionState()}
- * - catch up projections via {@see ContentRepository::catchUpProjection()}
+ * - send commands to the system (to mutate state) via {@see self::handle()}
+ * - access the content graph read model
+ * - access 3rd party read models via {@see self::projectionState()}
  *
  * @api
  */
-final class ContentRepository
+final readonly class ContentRepository
 {
-    /**
-     * @var array<class-string<ProjectionStateInterface>, ProjectionStateInterface>
-     */
-    private array $projectionStateCache;
-
     /**
      * @internal use the {@see ContentRepositoryFactory::getOrBuild()} to instantiate
      */
     public function __construct(
         public readonly ContentRepositoryId $id,
         private readonly CommandBus $commandBus,
-        private readonly EventStoreInterface $eventStore,
-        private readonly ProjectionsAndCatchUpHooks $projectionsAndCatchUpHooks,
-        private readonly EventNormalizer $eventNormalizer,
         private readonly EventPersister $eventPersister,
         private readonly NodeTypeManager $nodeTypeManager,
         private readonly InterDimensionalVariationGraph $variationGraph,
         private readonly ContentDimensionSourceInterface $contentDimensionSource,
         private readonly UserIdProviderInterface $userIdProvider,
         private readonly ClockInterface $clock,
-        private readonly ContentGraphReadModelInterface $contentGraphReadModel
+        private readonly ContentGraphReadModelInterface $contentGraphReadModel,
+        private readonly ProjectionStates $projectionStates,
     ) {
     }
 
@@ -118,120 +98,100 @@ public function handle(CommandInterface $command): void
      */
     public function projectionState(string $projectionStateClassName): ProjectionStateInterface
     {
-        if (!isset($this->projectionStateCache)) {
-            foreach ($this->projectionsAndCatchUpHooks->projections as $projection) {
-                if ($projection instanceof ContentGraphProjectionInterface) {
-                    continue;
-                }
-                $projectionState = $projection->getState();
-                $this->projectionStateCache[$projectionState::class] = $projectionState;
-            }
-        }
-        if (isset($this->projectionStateCache[$projectionStateClassName])) {
-            /** @var T $projectionState */
-            $projectionState = $this->projectionStateCache[$projectionStateClassName];
-            return $projectionState;
-        }
-        if (in_array(ContentGraphReadModelInterface::class, class_implements($projectionStateClassName), true)) {
-            throw new \InvalidArgumentException(sprintf('Accessing the internal content repository projection state via %s(%s) is not allowed. Please use the API on the content repository instead.', __FUNCTION__, $projectionStateClassName), 1729338679);
-        }
-
-        throw new \InvalidArgumentException(sprintf('A projection state of type "%s" is not registered in this content repository instance.', $projectionStateClassName), 1662033650);
-    }
-
-    /**
-     * @param class-string<ProjectionInterface<ProjectionStateInterface>> $projectionClassName
-     */
-    public function catchUpProjection(string $projectionClassName, CatchUpOptions $options): void
-    {
-        $projection = $this->projectionsAndCatchUpHooks->projections->get($projectionClassName);
-
-        $catchUpHookFactory = $this->projectionsAndCatchUpHooks->getCatchUpHookFactoryForProjection($projection);
-        $catchUpHook = $catchUpHookFactory?->build(new CatchUpHookFactoryDependencies(
-            $this->id,
-            $projection->getState(),
-            $this->nodeTypeManager,
-            $this->contentDimensionSource,
-            $this->variationGraph
-        ));
-
-        // TODO allow custom stream name per projection
-        $streamName = VirtualStreamName::all();
-        $eventStream = $this->eventStore->load($streamName);
-        if ($options->maximumSequenceNumber !== null) {
-            $eventStream = $eventStream->withMaximumSequenceNumber($options->maximumSequenceNumber);
-        }
-
-        $eventApplier = function (EventEnvelope $eventEnvelope) use ($projection, $catchUpHook, $options) {
-            $event = $this->eventNormalizer->denormalize($eventEnvelope->event);
-            if ($options->progressCallback !== null) {
-                ($options->progressCallback)($event, $eventEnvelope);
-            }
-            if (!$projection->canHandle($event)) {
-                return;
-            }
-            $catchUpHook?->onBeforeEvent($event, $eventEnvelope);
-            $projection->apply($event, $eventEnvelope);
-            if ($projection instanceof WithMarkStaleInterface) {
-                $projection->markStale();
-            }
-            $catchUpHook?->onAfterEvent($event, $eventEnvelope);
-        };
-
-        $catchUp = CatchUp::create($eventApplier, $projection->getCheckpointStorage());
-
-        if ($catchUpHook !== null) {
-            $catchUpHook->onBeforeCatchUp();
-            $catchUp = $catchUp->withOnBeforeBatchCompleted(fn() => $catchUpHook->onBeforeBatchCompleted());
+        try {
+            return $this->projectionStates->get($projectionStateClassName);
+        } catch (\InvalidArgumentException $e) {
+            throw new \InvalidArgumentException(sprintf('A projection state of type "%s" is not registered in this content repository instance: %s', $projectionStateClassName, $e->getMessage()), 1662033650, $e);
         }
-        $catchUp->run($eventStream);
-        $catchUpHook?->onAfterCatchUp();
     }
 
-    public function catchupProjections(): void
-    {
-        foreach ($this->projectionsAndCatchUpHooks->projections as $projection) {
-            // FIXME optimise by only loading required events once and not per projection
-            // see https://github.com/neos/neos-development-collection/pull/4988/
-            $this->catchUpProjection($projection::class, CatchUpOptions::create());
-        }
-    }
-
-    public function setUp(): void
-    {
-        $this->eventStore->setup();
-        foreach ($this->projectionsAndCatchUpHooks->projections as $projection) {
-            $projection->setUp();
-        }
-    }
-
-    public function status(): ContentRepositoryStatus
-    {
-        $projectionStatuses = ProjectionStatuses::createEmpty();
-        foreach ($this->projectionsAndCatchUpHooks->projections as $projectionClassName => $projection) {
-            $projectionStatuses = $projectionStatuses->with($projectionClassName, $projection->status());
-        }
-        return new ContentRepositoryStatus(
-            $this->eventStore->status(),
-            $projectionStatuses,
-        );
-    }
-
-    public function resetProjectionStates(): void
-    {
-        foreach ($this->projectionsAndCatchUpHooks->projections as $projection) {
-            $projection->reset();
-        }
-    }
-
-    /**
-     * @param class-string<ProjectionInterface<ProjectionStateInterface>> $projectionClassName
-     */
-    public function resetProjectionState(string $projectionClassName): void
-    {
-        $projection = $this->projectionsAndCatchUpHooks->projections->get($projectionClassName);
-        $projection->reset();
-    }
+//    /**
+//     * @param class-string<ProjectionInterface<ProjectionStateInterface>> $projectionClassName
+//     */
+//    public function catchUpProjection(string $projectionClassName, CatchUpOptions $options): void
+//    {
+//        $projection = $this->projectionsAndCatchUpHooks->projections->get($projectionClassName);
+//
+//        $catchUpHookFactory = $this->projectionsAndCatchUpHooks->getCatchUpHookFactoryForProjection($projection);
+//        $catchUpHook = $catchUpHookFactory?->build($this);
+//
+//        // TODO allow custom stream name per projection
+//        $streamName = VirtualStreamName::all();
+//        $eventStream = $this->eventStore->load($streamName);
+//        if ($options->maximumSequenceNumber !== null) {
+//            $eventStream = $eventStream->withMaximumSequenceNumber($options->maximumSequenceNumber);
+//        }
+//
+//        $eventApplier = function (EventEnvelope $eventEnvelope) use ($projection, $catchUpHook, $options) {
+//            $event = $this->eventNormalizer->denormalize($eventEnvelope->event);
+//            if ($options->progressCallback !== null) {
+//                ($options->progressCallback)($event, $eventEnvelope);
+//            }
+//            if (!$projection->canHandle($event)) {
+//                return;
+//            }
+//            $catchUpHook?->onBeforeEvent($event, $eventEnvelope);
+//            $projection->apply($event, $eventEnvelope);
+//            if ($projection instanceof WithMarkStaleInterface) {
+//                $projection->markStale();
+//            }
+//            $catchUpHook?->onAfterEvent($event, $eventEnvelope);
+//        };
+//
+//        $catchUp = CatchUp::create($eventApplier, $projection->getCheckpointStorage());
+//
+//        if ($catchUpHook !== null) {
+//            $catchUpHook->onBeforeCatchUp();
+//            $catchUp = $catchUp->withOnBeforeBatchCompleted(fn() => $catchUpHook->onBeforeBatchCompleted());
+//        }
+//        $catchUp->run($eventStream);
+//        $catchUpHook?->onAfterCatchUp();
+//    }
+
+//    public function catchupProjections(): void
+//    {
+//        foreach ($this->projectionsAndCatchUpHooks->projections as $projection) {
+//            // FIXME optimise by only loading required events once and not per projection
+//            // see https://github.com/neos/neos-development-collection/pull/4988/
+//            $this->catchUpProjection($projection::class, CatchUpOptions::create());
+//        }
+//    }
+
+//    public function setUp(): void
+//    {
+//        $this->eventStore->setup();
+//        foreach ($this->projectionsAndCatchUpHooks->projections as $projection) {
+//            $projection->setUp();
+//        }
+//    }
+
+//    public function status(): ContentRepositoryStatus
+//    {
+//        $projectionStatuses = ProjectionStatuses::createEmpty();
+//        foreach ($this->projectionsAndCatchUpHooks->projections as $projectionClassName => $projection) {
+//            $projectionStatuses = $projectionStatuses->with($projectionClassName, $projection->status());
+//        }
+//        return new ContentRepositoryStatus(
+//            $this->eventStore->status(),
+//            $projectionStatuses,
+//        );
+//    }
+
+//    public function resetProjectionStates(): void
+//    {
+//        foreach ($this->projectionsAndCatchUpHooks->projections as $projection) {
+//            $projection->reset();
+//        }
+//    }
+
+//    /**
+//     * @param class-string<ProjectionInterface<ProjectionStateInterface>> $projectionClassName
+//     */
+//    public function resetProjectionState(string $projectionClassName): void
+//    {
+//        $projection = $this->projectionsAndCatchUpHooks->projections->get($projectionClassName);
+//        $projection->reset();
+//    }
 
     /**
      * @throws WorkspaceDoesNotExist if the workspace does not exist

Neos.ContentRepository.Core/Classes/EventStore/EventPersister.php

@@ -39,7 +39,5 @@ public function publishEvents(ContentRepository $contentRepository, EventsToPubl
             $normalizedEvents,
             $eventsToPublish->expectedVersion
         );
-
-        $contentRepository->catchUpProjections();
     }
 }