feat: add change stream support

CHANGELOG.md

@@ -1,5 +1,6 @@
 # v8.3.0 (2024-09-02)
 
+- add support for change streams using Blueprint (#230)
 - add support for snapshot queries (#215)
 - deprecate Connection::getDatabaseContext() and move logic to UseMutations::getMutationExecutor() (#227)
 - add support for `Query\Builder::whereNotInUnnest(...)` (#225)

src/Schema/Blueprint.php

@@ -368,6 +368,32 @@ public function dropSequenceIfExists(string $name): Fluent
         return $this->addCommand(__FUNCTION__, ['sequence' => $name]);
     }
 
+    /**
+     * @param string $name
+     * @return ChangeStreamDefinition
+     */
+    public function createChangeStream(string $name): ChangeStreamDefinition
+    {
+        $this->commands[] = $command = new ChangeStreamDefinition(__FUNCTION__, $name);
+        return $command;
+    }
+
+    /**
+     * @param string $name
+     * @return ChangeStreamDefinition
+     */
+    public function alterChangeStream(string $name): ChangeStreamDefinition
+    {
+        $this->commands[] = $command = new ChangeStreamDefinition(__FUNCTION__, $name);
+        return $command;
+    }
+
+    public function dropChangeStream(string $name): ChangeStreamDefinition
+    {
+        $this->commands[] = $command = new ChangeStreamDefinition(__FUNCTION__, $name);
+        return $command;
+    }
+
     /**
      * @param string $type
      * @param string|list<string> $columns

src/Schema/ChangeStreamDefinition.php

@@ -0,0 +1,87 @@
+<?php
+
+/**
+ * Copyright 2019 Colopl Inc. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+declare(strict_types=1);
+
+namespace Colopl\Spanner\Schema;
+
+use Illuminate\Support\Fluent;
+
+/**
+ * @property string $retentionPeriod
+ * @property ChangeStreamValueCaptureType $valueCaptureType
+ * @property bool $excludeTtlDeletes
+ * @property bool $excludeInsert
+ * @property bool $excludeUpdate
+ * @property bool $excludeDelete
+ * @method $this retentionPeriod(string $retentionPeriod)
+ * @method $this valueCaptureType(ChangeStreamValueCaptureType $valueCaptureType)
+ * @method $this excludeTtlDeletes(bool $excludeTtlDeletes)
+ * @method $this excludeInsert(bool $excludeInsert)
+ * @method $this excludeUpdate(bool $excludeUpdate)
+ * @method $this excludeDelete(bool $excludeDelete)
+ * @extends Fluent<string, scalar>
+ */
+class ChangeStreamDefinition extends Fluent
+{
+    /**
+     * @param string $name
+     * @param string $stream
+     * @param array<string, list<string>> $tables
+     */
+    public function __construct(
+        public string $name,
+        public string $stream,
+        public array $tables = [],
+    ) {
+        parent::__construct();
+    }
+
+    /**
+     * @param string $table
+     * @param list<string> $columns
+     * @return $this
+     */
+    public function for(string $table, array $columns = []): static
+    {
+        $this->tables[$table] = $columns;
+        return $this;
+    }
+
+    /**
+     * @return array{
+     *     retentionPeriod?: string,
+     *     valueCaptureType?: ChangeStreamValueCaptureType,
+     *     excludeTtlDeletes?: bool,
+     *     excludeInsert?: bool,
+     *     excludeUpdate?: bool,
+     *     excludeDelete?: bool,
+     * }
+     */
+    public function getOptions(): array
+    {
+        return array_filter([
+            'retentionPeriod' => $this->retentionPeriod,
+            'valueCaptureType' => $this->valueCaptureType,
+            'excludeTtlDeletes' => $this->excludeTtlDeletes,
+            'excludeInsert' => $this->excludeInsert,
+            'excludeUpdate' => $this->excludeUpdate,
+            'excludeDelete' => $this->excludeDelete,
+        ], static fn($v) => $v !== null);
+    }
+}

src/Schema/ChangeStreamValueCaptureType.php

@@ -0,0 +1,29 @@
+<?php
+
+/**
+ * Copyright 2019 Colopl Inc. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+declare(strict_types=1);
+
+namespace Colopl\Spanner\Schema;
+
+enum ChangeStreamValueCaptureType: string
+{
+    case OldAndNewValues = 'OLD_AND_NEW_VALUES';
+    case NewValues = 'NEW_VALUES';
+    case NewRow = 'NEW_ROW';
+    case NewRowAndOldValues = 'NEW_ROW_AND_OLD_VALUES';
+}

src/Schema/Grammar.php

@@ -268,6 +268,84 @@ protected function formatSequenceOptions(mixed $definition): string
         return 'options (' . implode(', ', $optionAsStrings) . ')';
     }
 
+    /**
+     * @param Blueprint $blueprint
+     * @param ChangeStreamDefinition $command
+     * @return string
+     */
+    public function compileCreateChangeStream(Blueprint $blueprint, ChangeStreamDefinition $command): string
+    {
+        return implode(' ', array_filter([
+            "create change stream {$this->wrap($command->stream)}",
+            $this->formatChangeStreamTables($command),
+            $this->formatChangeStreamOptions($command),
+        ]));
+    }
+
+    /**
+     * @param Blueprint $blueprint
+     * @param ChangeStreamDefinition $command
+     * @return string
+     */
+    public function compileAlterChangeStream(Blueprint $blueprint, ChangeStreamDefinition $command): string
+    {
+        $parts = [];
+        $parts[] = "alter change stream {$this->wrap($command->stream)}";
+        if ($command->getOptions() !== []) {
+            $parts[] = 'set ' . $this->formatChangeStreamOptions($command);
+        }
+        return implode(' ', $parts);
+    }
+
+    /**
+     * @param Blueprint $blueprint
+     * @param ChangeStreamDefinition $command
+     * @return string
+     */
+    public function compileDropChangeStream(Blueprint $blueprint, ChangeStreamDefinition $command): string
+    {
+        return 'drop change stream ' . $this->wrap($command->stream);
+    }
+
+    protected function formatChangeStreamTables(ChangeStreamDefinition $definition): string
+    {
+        $parts = [];
+        foreach ($definition->tables as $table => $columns) {
+            $string = $this->wrap($table);
+            if ($columnsAsString = $this->columnize($columns)) {
+                $string .= "({$columnsAsString})";
+            }
+            $parts[] = $string;
+        }
+        return $parts !== []
+            ? 'for ' . implode(', ', $parts)
+            : 'for all';
+    }
+
+    /**
+     * @param ChangeStreamDefinition $definition
+     * @return string
+     */
+    protected function formatChangeStreamOptions(ChangeStreamDefinition $definition): string
+    {
+        $options = $definition->getOptions();
+
+        if ($options === []) {
+            return '';
+        }
+
+        $optionAsStrings = Arr::map($options, function (mixed $v, string $k): string {
+            return Str::snake($k) . '=' . match (true) {
+                $v === true => 'true',
+                $v === false => 'false',
+                is_string($v) => $this->quoteString($v),
+                $v instanceof ChangeStreamValueCaptureType => $this->quoteString($v->value),
+                default => throw new LogicException('Unsupported option value: ' . $v),
+            };
+        });
+        return 'options (' . implode(', ', $optionAsStrings) . ')';
+    }
+
     /**
      * @see https://cloud.google.com/spanner/docs/data-definition-language?hl=ja#create_table
      * @param Blueprint $blueprint

tests/Schema/BlueprintTest.php

@@ -18,6 +18,7 @@
 namespace Colopl\Spanner\Tests\Schema;
 
 use Colopl\Spanner\Schema\Blueprint;
+use Colopl\Spanner\Schema\ChangeStreamValueCaptureType;
 use Colopl\Spanner\Schema\Grammar;
 use Colopl\Spanner\Tests\TestCase;
 use Illuminate\Support\Arr;
@@ -513,6 +514,141 @@ public function test_dropSequenceIfExists(): void
         $blueprint->build($conn, $grammar);
     }
 
+    public function test_createChangeStream_for_all(): void
+    {
+        $conn = $this->getDefaultConnection();
+        $conn->useDefaultSchemaGrammar();
+        $grammar = $conn->getSchemaGrammar();
+        $name = 'test_stream' . Uuid::uuid4()->toString();
+
+        $blueprint = new Blueprint('_', fn(Blueprint $table) => $table->createChangeStream($name));
+        $blueprint->build($conn, $grammar);
+        $this->beforeApplicationDestroyed(fn() => $blueprint->dropChangeStream($name));
+
+        $this->assertSame(["create change stream `$name` for all"], $blueprint->toSql($conn, $grammar));
+        $this->assertContains($name, Arr::pluck($conn->select('SELECT * FROM INFORMATION_SCHEMA.CHANGE_STREAMS'), 'CHANGE_STREAM_NAME'));
+    }
+
+    public function test_createChangeStream_for_table(): void
+    {
+        $conn = $this->getDefaultConnection();
+        $conn->useDefaultSchemaGrammar();
+        $grammar = $conn->getSchemaGrammar();
+        $name = 'test_stream' . Uuid::uuid4()->toString();
+
+        $blueprint = new Blueprint('_', fn(Blueprint $table) => $table->createChangeStream($name)
+            ->for(self::TABLE_NAME_TEST)
+            ->for(self::TABLE_NAME_USER)
+        );
+        $blueprint->build($conn, $grammar);
+        $this->beforeApplicationDestroyed(fn() => $blueprint->dropChangeStream($name));
+
+        $this->assertSame(["create change stream `{$name}` for `Test`, `User`"], $blueprint->toSql($conn, $grammar));
+    }
+
+    public function test_createChangeStream_for_table_columns(): void
+    {
+        $conn = $this->getDefaultConnection();
+        $conn->useDefaultSchemaGrammar();
+        $grammar = $conn->getSchemaGrammar();
+        $name = 'test_stream' . Uuid::uuid4()->toString();
+
+        $blueprint = new Blueprint('_', fn(Blueprint $table) => $table->createChangeStream($name)->for(self::TABLE_NAME_TEST, ['stringTest', 'intTest']));
+        $blueprint->build($conn, $grammar);
+        $this->beforeApplicationDestroyed(fn() => $blueprint->dropChangeStream($name));
+
+        $this->assertSame(["create change stream `$name` for `Test`(`stringTest`, `intTest`)"], $blueprint->toSql($conn, $grammar));
+    }
+
+    public function test_createChangeStream_for_with_options(): void
+    {
+        $conn = $this->getDefaultConnection();
+        $conn->useDefaultSchemaGrammar();
+        $grammar = $conn->getSchemaGrammar();
+        $uuid = Uuid::uuid4()->toString();
+        $tableName = self::TABLE_NAME_TEST . $uuid;
+        $streamName = 'test_stream' . $uuid;
+
+        $blueprint = new Blueprint($tableName, function (Blueprint $table) use ($streamName) {
+            $table->uuid('id')->primary();
+            $table->create();
+            $table->createChangeStream($streamName)
+                ->for($table->getTable())
+                ->excludeTtlDeletes(true)
+                ->excludeInsert(true)
+                ->excludeUpdate(true)
+                ->excludeDelete(true)
+                ->valueCaptureType(ChangeStreamValueCaptureType::NewRow)
+                ->retentionPeriod('1d');
+        });
+        $blueprint->build($conn, $grammar);
+        $this->beforeApplicationDestroyed(fn() => $blueprint->dropChangeStream($name));
+
+        $this->assertSame([
+            "create table `$tableName` (`id` string(36) not null) primary key (`id`)",
+            "create change stream `$streamName` for `$tableName` options (" . implode(', ', [
+                "retention_period='1d'",
+                "value_capture_type='NEW_ROW'",
+                "exclude_ttl_deletes=true",
+                "exclude_insert=true",
+                "exclude_update=true",
+                "exclude_delete=true",
+            ]) . ")",
+        ], $blueprint->toSql($conn, $grammar));
+    }
+
+    public function test_alterChangeStream(): void
+    {
+        $conn = $this->getDefaultConnection();
+        $conn->useDefaultSchemaGrammar();
+        $grammar = $conn->getSchemaGrammar();
+        $uuid = Uuid::uuid4()->toString();
+        $tableName = self::TABLE_NAME_TEST . $uuid;
+        $streamName = 'test_stream' . $uuid;
+
+        $blueprint = new Blueprint($tableName, function (Blueprint $table) use ($streamName) {
+            $table->uuid('id')->primary();
+            $table->create();
+            $table->createChangeStream($streamName)->for($table->getTable())->excludeTtlDeletes(true);
+        });
+        $blueprint->build($conn, $grammar);
+        $this->beforeApplicationDestroyed(fn() => $blueprint->dropChangeStream($streamName));
+
+        $blueprint = new Blueprint('_', function (Blueprint $table) use ($streamName) {
+            $table->alterChangeStream($streamName)
+                ->excludeTtlDeletes(false)
+                ->retentionPeriod('7d');
+        });
+        $blueprint->build($conn, $grammar);
+
+        $this->assertSame([
+            "alter change stream `$streamName` set options (retention_period='7d', exclude_ttl_deletes=false)",
+        ], $blueprint->toSql($conn, $grammar));
+    }
+
+    public function test_dropChangeStream(): void
+    {
+        $conn = $this->getDefaultConnection();
+        $conn->useDefaultSchemaGrammar();
+        $grammar = $conn->getSchemaGrammar();
+        $uuid = Uuid::uuid4()->toString();
+        $streamName = 'test_stream' . $uuid;
+
+        $blueprint = new Blueprint(self::TABLE_NAME_TEST . $uuid, function (Blueprint $table) use ($streamName) {
+            $table->uuid('id')->primary();
+            $table->create();
+            $table->createChangeStream($streamName)->for($table->getTable());
+        });
+        $blueprint->build($conn, $grammar);
+
+        $blueprint = new Blueprint('_', fn (Blueprint $table) => $table->dropChangeStream($streamName));
+        $blueprint->build($conn, $grammar);
+
+        $this->assertSame([
+            "drop change stream `$streamName`",
+        ], $blueprint->toSql($conn, $grammar));
+    }
+
     public function test_default_values(): void
     {
         $conn = $this->getDefaultConnection();