Daikon\CouchDb\Storage\CouchDbStorageAdapter

Complexity

Total Complexity

14

Size/Duplication

Total Lines	85
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	5

Importance

Changes

0

6 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	5	1
B	read()	0	36	5
A	write()	0	9	3
A	delete()	0	4	1
A	request()	0	10	2
A	buildUri()	0	9	2

<?php

namespace Daikon\CouchDb\Storage;

use Daikon\CouchDb\Connector\CouchDbConnector;
use Daikon\Dbal\Exception\DbalException;
use Daikon\Dbal\Storage\StorageAdapterInterface;
use Daikon\EventSourcing\EventStore\CommitSequence;
use GuzzleHttp\Exception\RequestException;
use GuzzleHttp\Psr7\Request;

final class CouchDbStorageAdapter implements StorageAdapterInterface
{
    private $connector;

    private $settings;

    public function __construct(CouchDbConnector $connector, array $settings = [])
    {
        $this->connector = $connector;
        $this->settings = $settings;
    }

    public function read(string $identifier)
    {
        $viewPath = sprintf(
            '/_design/%s/_view/%s',
            $this->settings['design_doc'],
            $this->settings['view_name'] ?? 'commit_stream'
        );

        $viewParams = [
            'startkey' => sprintf('["%s", {}]', $identifier),
            'endkey' => sprintf('["%s", 1]', $identifier),
            'include_docs' => 'true',
            'reduce' => 'false',
            'descending' => 'true',
            'limit' => 1000 // @todo use snapshot size config setting as soon as available
        ];

        try {
            $response = $this->request($viewPath, 'GET', [], $viewParams);
            $rawResponse = json_decode($response->getBody(), true);
        } catch (RequestException $error) {
            if ($error->hasResponse() && $error->getResponse()->getStatusCode() === 404) {
                return CommitSequence::makeEmpty();
            } else {
                throw $error;
            }
        }

        if (!isset($rawResponse['total_rows'])) {
            throw new DbalException('Failed to read data for '.$identifier);
        }

        return CommitSequence::fromArray(array_map(function (array $commitData) {
            return $commitData['doc'];
        }, array_reverse($rawResponse['rows'])));
    }

    public function write(string $identifier, array $body)
    {
        $response = $this->request($identifier, 'PUT', $body);
        $rawResponse = json_decode($response->getBody(), true);

        if (!isset($rawResponse['ok']) || !isset($rawResponse['rev'])) {
            throw new DbalException('Failed to write data for '.$identifier);
        }
    }

    public function delete(string $identifier)
    {
        throw new DbalException('Not yet implemented');
    }

    private function request(string $identifier, string $method, array $body = [], array $params = [])
    {
        $uri = $this->buildUri($identifier, $params);

        $request = empty($body)
            ? new Request($method, $uri)
            : new Request($method, $uri, [], json_encode($body));

        return $this->connector->getConnection()->send($request);
    }

    private function buildUri(string $identifier, array $params = [])
    {
        $settings = $this->connector->getSettings();
        $uri = sprintf('/%s/%s', $settings['database'], $identifier);
        if (!empty($params)) {
            $uri .= '?'.http_build_query($params);
        }
        return str_replace('//', '/', $uri);
    }
}