From 20316d040ddd25c522bb073be888f121b913ff40 Mon Sep 17 00:00:00 2001
From: Daniel Subiabre <daniel.subiabre@platoniq.net>
Date: Thu, 14 Mar 2024 16:59:51 +0100
Subject: [PATCH] Start benzina db streaming tool

---
 src/Library/Benzina/Pdo/PdoReader.php   | 46 ++++++++++++++
 src/Library/Benzina/Pdo/PdoStream.php   | 48 +++++++++++++++
 src/Library/Benzina/ReaderInterface.php |  8 +++
 src/Library/Benzina/Stream.php          | 82 +++++++++++++++++++++++++
 src/Library/Benzina/StreamInterface.php | 17 +++++
 src/Library/Benzina/WriterInterface.php |  8 +++
 6 files changed, 209 insertions(+)
 create mode 100644 src/Library/Benzina/Pdo/PdoReader.php
 create mode 100644 src/Library/Benzina/Pdo/PdoStream.php
 create mode 100644 src/Library/Benzina/ReaderInterface.php
 create mode 100644 src/Library/Benzina/Stream.php
 create mode 100644 src/Library/Benzina/StreamInterface.php
 create mode 100644 src/Library/Benzina/WriterInterface.php

diff --git a/src/Library/Benzina/Pdo/PdoReader.php b/src/Library/Benzina/Pdo/PdoReader.php
new file mode 100644
index 0000000..867858a
--- /dev/null
+++ b/src/Library/Benzina/Pdo/PdoReader.php
@@ -0,0 +1,46 @@
+<?php
+
+namespace App\Library\Benzina\Pdo;
+
+use App\Library\Benzina\ReaderInterface;
+use App\Library\Benzina\Stream;
+
+class PdoReader implements ReaderInterface
+{
+    private \PDO $db;
+
+    /**
+     * @var array{name: string, schema: string, host: string, user: string, pass: string}
+     */
+    public readonly array $config;
+
+    public function __construct(string $databaseUrl)
+    {
+        $parsedUrl = parse_url($databaseUrl);
+        $dbdata = [
+            'name' => ltrim($parsedUrl['path'], '/'),
+            ...$parsedUrl
+        ];
+
+        $this->db = new \PDO(
+            dsn: sprintf("%s:host=%s;dbname=%s", $dbdata['scheme'], $dbdata['host'], $dbdata['name']),
+            username: $dbdata['user'],
+            password: $dbdata['pass'],
+            options: [
+                \PDO::ATTR_EMULATE_PREPARES   => false,
+                \PDO::ATTR_DEFAULT_FETCH_MODE => \PDO::FETCH_ASSOC
+            ]
+        );
+
+        $this->config = $dbdata;
+    }
+
+    public function get(string $name): PdoStream
+    {
+        $count = $this->db->query("SELECT COUNT(*) FROM `$name`;")->fetchColumn();
+        $query = $this->db->prepare("SELECT * FROM `$name` LIMIT ? OFFSET ?;");
+
+        $stream = new Stream(str_repeat("\n", $count));
+        return new PdoStream($stream, $query);
+    }
+}
diff --git a/src/Library/Benzina/Pdo/PdoStream.php b/src/Library/Benzina/Pdo/PdoStream.php
new file mode 100644
index 0000000..5d906d7
--- /dev/null
+++ b/src/Library/Benzina/Pdo/PdoStream.php
@@ -0,0 +1,48 @@
+<?php
+
+namespace App\Library\Benzina\Pdo;
+
+use App\Library\Benzina\Stream;
+use App\Library\Benzina\StreamInterface;
+
+class PdoStream implements StreamInterface
+{
+    private const BATCH_SIZE = 99;
+    private int $currentBatch = 0;
+
+    public function __construct(
+        private Stream $stream,
+        private \PDOStatement $pdo,
+    ) {
+    }
+
+    public function eof(): bool
+    {
+        return $this->stream->eof();
+    }
+
+    public function read(?int $length = null): mixed
+    {
+        while (!$this->stream->eof()) {
+            $this->stream->read(self::BATCH_SIZE);
+            $this->pdo->execute([self::BATCH_SIZE, $this->currentBatch]);
+
+            $result = $this->pdo->fetchAll();
+            $this->currentBatch += self::BATCH_SIZE;
+
+            return $result;
+        }
+
+        $this->stream->close();
+    }
+
+    public function close(): void
+    {
+        $this->stream->close();
+    }
+
+    public function length(): int
+    {
+        return $this->stream->length();
+    }
+}
diff --git a/src/Library/Benzina/ReaderInterface.php b/src/Library/Benzina/ReaderInterface.php
new file mode 100644
index 0000000..daa8259
--- /dev/null
+++ b/src/Library/Benzina/ReaderInterface.php
@@ -0,0 +1,8 @@
+<?php
+
+namespace App\Library\Benzina;
+
+interface ReaderInterface
+{
+    public function get(string $entityName): StreamInterface;
+}
diff --git a/src/Library/Benzina/Stream.php b/src/Library/Benzina/Stream.php
new file mode 100644
index 0000000..7e50595
--- /dev/null
+++ b/src/Library/Benzina/Stream.php
@@ -0,0 +1,82 @@
+<?php
+
+namespace App\Library\Benzina;
+
+class Stream implements StreamInterface
+{
+    /** @var resource */
+    private $stream;
+
+    public function __construct(string $data)
+    {
+        try {
+            $this->stream = \fopen('php://temp', 'r+');
+
+            \fwrite($this->stream, $data);
+            \fseek($this->stream, 0);
+        } catch (\Exception $e) {
+            throw new \RuntimeException("Unable to open stream");
+        }
+    }
+
+    public function __destruct()
+    {
+        $this->close();
+    }
+
+    public function eof(): bool
+    {
+        if (!isset($this->stream)) {
+            throw new \RuntimeException(self::MESSAGE_ERROR_DETACHED);
+        }
+
+        return \feof($this->stream);
+    }
+
+    public function read(?int $length = null): string
+    {
+        if (!isset($this->stream)) {
+            throw new \RuntimeException(self::MESSAGE_ERROR_DETACHED);
+        }
+
+        if ($length < 0) {
+            throw new \RuntimeException("Length parameter cannot be negative");
+        }
+
+        if ($length === 0) {
+            return '';
+        }
+
+        try {
+            $string = \fgets($this->stream, $length);
+        } catch (\Exception $e) {
+            throw new \RuntimeException(self::MESSAGE_ERROR_UNABLE_TO_READ, 0, $e);
+        }
+
+        if ($string === false) {
+            throw new \RuntimeException(self::MESSAGE_ERROR_UNABLE_TO_READ);
+        }
+
+        return $string;
+    }
+
+    public function close(): void
+    {
+        if (isset($this->stream)) {
+            if (\is_resource($this->stream)) {
+                \fclose($this->stream);
+            }
+
+            unset($this->stream);
+        }
+    }
+
+    public function length(): int
+    {
+        if (!isset($this->stream)) {
+            throw new \RuntimeException(self::MESSAGE_ERROR_DETACHED);
+        }
+
+        return \fstat($this->stream)['size'];
+    }
+}
diff --git a/src/Library/Benzina/StreamInterface.php b/src/Library/Benzina/StreamInterface.php
new file mode 100644
index 0000000..2513990
--- /dev/null
+++ b/src/Library/Benzina/StreamInterface.php
@@ -0,0 +1,17 @@
+<?php
+
+namespace App\Library\Benzina;
+
+interface StreamInterface
+{
+    public const MESSAGE_ERROR_DETACHED = 'Stream is detached';
+    public const MESSAGE_ERROR_UNABLE_TO_READ = 'Unable to read from stream';
+
+    public function eof(): bool;
+
+    public function read(?int $length = null): mixed;
+
+    public function close(): void;
+
+    public function length(): int;
+}
diff --git a/src/Library/Benzina/WriterInterface.php b/src/Library/Benzina/WriterInterface.php
new file mode 100644
index 0000000..2f063bf
--- /dev/null
+++ b/src/Library/Benzina/WriterInterface.php
@@ -0,0 +1,8 @@
+<?php
+
+namespace App\Library\Benzina;
+
+interface WriterInterface
+{
+    public function process(ReaderInterface $reader);
+}