litebase-php/src/HttpStreamingTransport.php at 84f00f497c8154e9e3a648d7cc805f1af5b7ca3f · litebase/litebase-php · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
<?php

namespace Litebase;

use Exception;
use Litebase\Exceptions\LitebaseConnectionException;

class HttpStreamingTransport implements TransportInterface
{
    use HasRequestHeaders;
    use SignsRequests;

    protected Connection $connection;

    /**
     * Create a new instance of the transport.
     */
    public function __construct(
        protected Configuration $config,
    ) {}

    public function send(Query $query): ?QueryResult
    {
        $path = sprintf(
            'v1/databases/%s/branches/%s/query/stream',
            $this->config->getDatabase(),
            $this->config->getBranch()
        );

        if (! isset($this->connection) || ! $this->connection->isOpen()) {
            $headers = $this->requestHeaders(
                host: $this->config->getHost(),
                port: $this->config->getPort(),
                contentLength: 0,
                headers: [
                    'Content-Type' => 'application/octet-stream',
                ]
            );

            $url = $this->config->getPort() === null
                ? sprintf('https://%s/%s', $this->config->getHost(), $path)
                : sprintf('http://%s:%d/%s', $this->config->getHost(), $this->config->getPort(), $path);

            if (! empty($this->config->getUsername()) || ! (empty($this->config->getPassword()))) {
                $headers['Authorization'] = 'Basic '.base64_encode($this->config->getUsername().':'.$this->config->getPassword());
            }

            if (! empty($this->config->getAccessToken())) {
                $headers['Authorization'] = 'Bearer '.$this->config->getAccessToken();
            }

            if (! empty($this->config->getAccessKeyId())) {
                $token = $this->getToken(
                    accessKeyID: $this->config->getAccessKeyId(),
                    accessKeySecret: $this->config->getAccessKeySecret(),
                    method: 'POST',
                    path: $path,
                    headers: $headers,
                    data: null,
                );

                $headers['Authorization'] = sprintf('Litebase-HMAC-SHA256 %s', $token);
            }

            $this->connection = new Connection($url, $headers);
        }

        try {
            $result = $this->connection->send($query);
        } catch (Exception $e) {
            throw new LitebaseConnectionException(
                code: $e->getCode(),
                message: $e->getMessage(),
            );
        }

        // if ($result === null) {
        //     $this->connection->close();

        //     throw new LitebaseConnectionClosedException(
        //         code: 0,
        //         message: 'Connection closed',
        //     );
        // }

        if (($result->errorMessage ?? null) === 'error') {
            $this->connection->close();

            throw new LitebaseConnectionException(
                code: $result->errorCode ?? 0,
                message: $result->errorMessage ?? 'Unknown error',
            );
        }

        return $result;
    }
}