SmoothPhp\LaravelAdapter\EventStore\LaravelEventStore

Complexity

Total Complexity

14

Size/Duplication

Total Lines	151
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	10

Importance

Changes	1
Bugs	0	Features	0

<?php
 */
final class LaravelEventStore implements EventStore
{
    /** @var Serializer */
    private $serializer;

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

    /** @var Connection */
    private $db;

    /**
     * @param DatabaseManager $databaseManager
     * @param Serializer $serializer
     * @param string $eventStoreConnectionName
     * @param string $eventStoreTableName
     */
    public function __construct(
        DatabaseManager $databaseManager,
        Serializer $serializer,
        $eventStoreConnectionName,
        $eventStoreTableName
    ) {
        $this->db = $databaseManager->connection($eventStoreConnectionName);
        $this->serializer = $serializer;
        $this->eventStoreTableName = $eventStoreTableName;
    }

    /**
     * @param string $id
     * @return DomainEventStream
     * @throws EventStreamNotFound
     */
    public function load($id)
    {
        $rows = $this->db->table($this->eventStoreTableName)
                         ->select(['uuid', 'playhead', 'metadata', 'payload', 'recorded_on'])
                         ->where('uuid', $id)
                         ->orderBy('playhead', 'asc')
                         ->get();
        $events = [];

        foreach ($rows as $row) {
            $events[] = $this->deserializeEvent($row);
        }

        if (empty($events)) {
            throw new EventStreamNotFound(sprintf('EventStream not found for aggregate with id %s', $id));
        }

        return new \SmoothPhp\Domain\DomainEventStream($events);
    }

    /**
     * @param mixed $id
     * @param DomainEventStream $eventStream
     */
    public function append($id, DomainEventStream $eventStream)
    {
        $id = (string)$id; //Used to thrown errors if ID will not cast to string

        $this->db->beginTransaction();

        try {
            foreach ($eventStream as $domainMessage) {
                $this->insertEvent($domainMessage);
            }

            $this->db->commit();
        } catch (QueryException $ex) {
            $this->db->rollBack();

            throw  $ex;
        }
    }

    /**
     * @param DomainMessage $domainMessage
     */
    private function insertEvent(DomainMessage $domainMessage)
    {
        try {
            $this->db->table($this->eventStoreTableName)
                     ->insert(
                         [
                             'uuid'        => (string)$domainMessage->getId(),
                             'playhead'    => $domainMessage->getPlayHead(),
                             'metadata'    => json_encode($this->serializer->serialize($domainMessage->getMetadata())),
                             'payload'     => json_encode($this->serializer->serialize($domainMessage->getPayload())),
                             'recorded_on' => (string)$domainMessage->getRecordedOn(),
                             'type'        => $domainMessage->getType(),
                         ]
                     );
        } catch (\PDOException $ex) {
            if ($ex->getCode() == 23000) {
                throw new DuplicateAggregatePlayhead((string)$domainMessage->getId(), $domainMessage->getPlayHead());
            }
            throw  $ex;
        }
    }

    /**
     * @param \stdClass
     * @return DomainMessage
     */
    private function deserializeEvent($row)
    {
        return new \SmoothPhp\Domain\DomainMessage(
            $row->uuid,
            $row->playhead,
            $this->serializer->deserialize(json_decode($row->metadata, true)),
            $this->serializer->deserialize(json_decode($row->payload, true)),
            new DateTime($row->recorded_on)
        );
    }

    /**
     * @param string[] $eventTypes
     * @return int
     */
    public function getEventCountByTypes($eventTypes)
    {
        return $this->db->table($this->eventStoreTableName)
                        ->whereIn('type', $eventTypes)
                        ->count();
    }

    /**
     * @param string[] $eventTypes
     * @param int $skip
     * @param int $take
     * @return DomainEventStream
     */
    public function getEventsByType($eventTypes, $skip, $take)
    {
        $rows = $this->db->table($this->eventStoreTableName)
                         ->select(['uuid', 'playhead', 'metadata', 'payload', 'recorded_on'])
                         ->whereIn('type', $eventTypes)
                         ->skip($skip)
                         ->take($take)
                         ->orderBy('recorded_on', 'asc')
                         ->get();
        $events = [];

        foreach ($rows as $row) {
            $events[] = $this->deserializeEvent($row);
        }

        return new \SmoothPhp\Domain\DomainEventStream($events);
    }
}