Daikon\EventSourcing\EventStore\UnitOfWork

Complexity

Total Complexity

20

Size/Duplication

Total Lines	122
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	11

Test Coverage

Coverage

39.34%

Importance

Changes

0

6 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	14	1
A	commit()	0	21	4
A	checkout()	0	13	1
A	getTrackedStream()	0	14	3
B	buildEventHistory()	0	16	6
B	detectConflictingEvents()	0	16	5

<?php
/**
 * This file is part of the daikon-cqrs/cqrs project.
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

declare(strict_types=1);

namespace Daikon\EventSourcing\EventStore;

use Daikon\EventSourcing\Aggregate\AggregateIdInterface;
use Daikon\EventSourcing\Aggregate\AggregateRevision;
use Daikon\EventSourcing\Aggregate\AggregateRootInterface;
use Daikon\EventSourcing\Aggregate\DomainEventSequence;
use Daikon\MessageBus\Metadata\Metadata;

final class UnitOfWork implements UnitOfWorkInterface
{
    private const MAX_RESOLUTION_ATTEMPTS = 5;

    /** @var string */
    private $aggregateRootType;

    /** @var StreamStoreInterface */
    private $streamStore;

    /** @var StreamProcessorInterface */
    private $streamProcessor;

    /** @var string */
    private $streamImplementor;

    /** @var StreamMap */
    private $trackedCommitStreams;

    /** @var int */
    private $maxResolutionAttempts;

    public function __construct(
        string $aggregateRootType,
        StreamStoreInterface $streamStore,
        StreamProcessorInterface $streamProcessor = null,
        string $streamImplementor = Stream::class,
        int $maxResolutionAttempts = self::MAX_RESOLUTION_ATTEMPTS
    ) {
        $this->aggregateRootType = $aggregateRootType;
        $this->streamStore = $streamStore;
        $this->streamProcessor = $streamProcessor;
        $this->streamImplementor = $streamImplementor;
        $this->trackedCommitStreams = StreamMap::makeEmpty();

It seems like \Daikon\EventSourcing\EventStore\StreamMap::makeEmpty() of type object<self> is incompatible with the declared type object<Daikon\EventSourcing\EventStore\StreamMap> of property $trackedCommitStreams.

        $this->maxResolutionAttempts = $maxResolutionAttempts;
    }

    public function commit(AggregateRootInterface $aggregateRoot, Metadata $metadata): CommitSequence
    {
        $prevStream = $this->getTrackedStream($aggregateRoot);
        $updatedStream = $prevStream->appendEvents($aggregateRoot->getTrackedEvents(), $metadata);
        $result = $this->streamStore->commit($updatedStream, $prevStream->getStreamRevision());
        $resolutionAttempts = 0;
        while ($result instanceof PossibleConflict) {
            $conflictingStream = $this->streamStore->checkout($updatedStream->getStreamId());
            $conflictingEvents = $this->detectConflictingEvents($aggregateRoot, $conflictingStream);
            if (!$conflictingEvents->isEmpty()) {
                throw new UnresolvableConflict($conflictingStream->getStreamId(), $conflictingEvents);
            }
            $resolvedStream = $conflictingStream->appendEvents($aggregateRoot->getTrackedEvents(), $metadata);
            $result = $this->streamStore->commit($resolvedStream, $conflictingStream->getStreamRevision());
            if (++$resolutionAttempts >= $this->maxResolutionAttempts) {
                throw new ConcurrencyRaceLost($conflictingStream->getStreamId(), $aggregateRoot->getTrackedEvents());
            }
        }
        $this->trackedCommitStreams = $this->trackedCommitStreams->unregister($prevStream);
        return $updatedStream->getCommitRange($prevStream->getStreamRevision(), $updatedStream->getStreamRevision());
    }

    public function checkout(AggregateIdInterface $aggregateId, AggregateRevision $revision): AggregateRootInterface
    {
        /** @var $streamId StreamId */
        $streamId = StreamId::fromNative($aggregateId->toNative());
        $stream = $this->streamStore->checkout($streamId, $revision);
        $aggregateRoot = call_user_func(
            [ $this->aggregateRootType, 'reconstituteFromHistory' ],
            $aggregateId,
            $this->buildEventHistory($stream, $revision)
        );
        $this->trackedCommitStreams = $this->trackedCommitStreams->register($stream);
        return $aggregateRoot;
    }

    private function getTrackedStream(AggregateRootInterface $aggregateRoot): StreamInterface
    {
        $streamId = StreamId::fromNative((string)$aggregateRoot->getIdentifier());
        $tailRevision = $aggregateRoot->getTrackedEvents()->getTailRevision();
        if ($this->trackedCommitStreams->has((string)$streamId)) {
            $stream = $this->trackedCommitStreams->get((string)$streamId);
        } elseif ($tailRevision->isInitial()) {
            $stream = call_user_func([ $this->streamImplementor, 'fromStreamId' ], $streamId);
            $this->trackedCommitStreams = $this->trackedCommitStreams->register($stream);
        } else {
            throw new \Exception("AggregateRoots must be checked out before they may be committed.");
        }
        return $stream;
    }

    private function buildEventHistory(StreamInterface $stream, AggregateRevision $to): DomainEventSequence
    {
        $history = DomainEventSequence::makeEmpty();
        if ($this->streamProcessor) {
            $stream = $this->streamProcessor->process($stream);
        }
        foreach ($stream as $commit) {
            if (!$to->isEmpty() && $commit->getAggregateRevision()->isGreaterThan($to)) {
                break;
            }
            foreach ($commit->getEventLog() as $event) {
                $history = $history->push($event);
            }
        }
        return $history;
    }

    private function detectConflictingEvents(
        AggregateRootInterface $aggregateRoot,
        StreamInterface $stream
    ): DomainEventSequence {
        $conflictingEvents = DomainEventSequence::makeEmpty();
        foreach ($aggregateRoot->getTrackedEvents() as $newEvent) {
            foreach ($stream->findCommitsSince($aggregateRoot->getRevision()) as $previousCommit) {
                foreach ($previousCommit->getEventLog() as $previousEvent) {
                    if ($newEvent->conflictsWith($previousEvent)) {
                        $conflictingEvents = $conflictingEvents->push($newEvent);
                    }
                }
            }
        }
        return $conflictingEvents;
    }
}