From 55e10ecff2646ef63fbb1a40a3ecc9eff12e0573 Mon Sep 17 00:00:00 2001
From: David Badura <david.badura@patchlevel.de>
Date: Wed, 31 Jul 2024 08:12:00 +0200
Subject: [PATCH] add subscription setup cli commands & skip setup in
 subscription run

---
 .../SchemaSubscriptionSetupCommand.php        | 47 ++++++++++++++++
 .../SchemaSubscriptionTeardownCommand.php     | 55 +++++++++++++++++++
 .../Command/SubscriptionRunCommand.php        |  9 ++-
 src/Store/DoctrineDbalStore.php               | 19 +++++++
 src/Store/StreamDoctrineDbalStore.php         | 19 +++++++
 src/Store/SubscriptionStore.php               |  2 +
 6 files changed, 150 insertions(+), 1 deletion(-)
 create mode 100644 src/Console/Command/SchemaSubscriptionSetupCommand.php
 create mode 100644 src/Console/Command/SchemaSubscriptionTeardownCommand.php

diff --git a/src/Console/Command/SchemaSubscriptionSetupCommand.php b/src/Console/Command/SchemaSubscriptionSetupCommand.php
new file mode 100644
index 000000000..3be8e456e
--- /dev/null
+++ b/src/Console/Command/SchemaSubscriptionSetupCommand.php
@@ -0,0 +1,47 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Console\Command;
+
+use Patchlevel\EventSourcing\Console\OutputStyle;
+use Patchlevel\EventSourcing\Store\Store;
+use Patchlevel\EventSourcing\Store\SubscriptionStore;
+use Symfony\Component\Console\Attribute\AsCommand;
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+
+#[AsCommand(
+    'event-sourcing:schema:subscription-setup',
+    'setup subscription (pub/sub) for store',
+)]
+final class SchemaSubscriptionSetupCommand extends Command
+{
+    public function __construct(
+        private readonly Store $store,
+    ) {
+        parent::__construct();
+    }
+
+    protected function execute(InputInterface $input, OutputInterface $output): int
+    {
+        $io = new OutputStyle($input, $output);
+
+        if (!$this->store instanceof SubscriptionStore) {
+            $io->error('store does not support subscriptions');
+
+            return 1;
+        }
+
+        if (!$this->store->supportSubscription()) {
+            $io->error('store does not support subscriptions');
+
+            return 1;
+        }
+
+        $this->store->setupSubscription();
+
+        return 0;
+    }
+}
diff --git a/src/Console/Command/SchemaSubscriptionTeardownCommand.php b/src/Console/Command/SchemaSubscriptionTeardownCommand.php
new file mode 100644
index 000000000..4571d7b1b
--- /dev/null
+++ b/src/Console/Command/SchemaSubscriptionTeardownCommand.php
@@ -0,0 +1,55 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Console\Command;
+
+use Patchlevel\EventSourcing\Console\OutputStyle;
+use Patchlevel\EventSourcing\Store\Store;
+use Patchlevel\EventSourcing\Store\SubscriptionStore;
+use Symfony\Component\Console\Attribute\AsCommand;
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+
+use function method_exists;
+
+#[AsCommand(
+    'event-sourcing:schema:subscription-teardown',
+    'teardown subscription (pub/sub) for store',
+)]
+final class SchemaSubscriptionTeardownCommand extends Command
+{
+    public function __construct(
+        private readonly Store $store,
+    ) {
+        parent::__construct();
+    }
+
+    protected function execute(InputInterface $input, OutputInterface $output): int
+    {
+        $io = new OutputStyle($input, $output);
+
+        if (!$this->store instanceof SubscriptionStore) {
+            $io->error('store does not support subscriptions');
+
+            return 1;
+        }
+
+        if (!$this->store->supportSubscription()) {
+            $io->error('store does not support subscriptions');
+
+            return 1;
+        }
+
+        if (method_exists($this->store, 'teardownSubscription')) {
+            $this->store->teardownSubscription();
+
+            return 0;
+        }
+
+        $io->error('store does not support teardownSubscription');
+
+        return 1;
+    }
+}
diff --git a/src/Console/Command/SubscriptionRunCommand.php b/src/Console/Command/SubscriptionRunCommand.php
index 87548d907..a9a33a6bc 100644
--- a/src/Console/Command/SubscriptionRunCommand.php
+++ b/src/Console/Command/SubscriptionRunCommand.php
@@ -70,6 +70,12 @@ protected function configure(): void
                 null,
                 InputOption::VALUE_NONE,
                 'rebuild (remove & boot) subscriptions before run',
+            )
+            ->addOption(
+                'skip-subscription-setup',
+                null,
+                InputOption::VALUE_NONE,
+                'skip subscription setup',
             );
     }
 
@@ -81,11 +87,12 @@ protected function execute(InputInterface $input, OutputInterface $output): int
         $timeLimit = InputHelper::nullablePositiveInt($input->getOption('time-limit'));
         $sleep = InputHelper::positiveIntOrZero($input->getOption('sleep'));
         $rebuild = InputHelper::bool($input->getOption('rebuild'));
+        $skipSubscriptionSetup = InputHelper::bool($input->getOption('skip-subscription-setup'));
 
         $criteria = $this->subscriptionEngineCriteria($input);
         $criteria = $this->resolveCriteriaIntoCriteriaWithOnlyIds($criteria);
 
-        if ($this->store instanceof SubscriptionStore) {
+        if ($this->store instanceof SubscriptionStore && !$skipSubscriptionSetup) {
             $this->store->setupSubscription();
         }
 
diff --git a/src/Store/DoctrineDbalStore.php b/src/Store/DoctrineDbalStore.php
index 2a4104274..4a45b7c61 100644
--- a/src/Store/DoctrineDbalStore.php
+++ b/src/Store/DoctrineDbalStore.php
@@ -410,6 +410,25 @@ public function setupSubscription(): void
         ));
     }
 
+    public function teardownSubscription(): void
+    {
+        if (!$this->supportSubscription()) {
+            return;
+        }
+
+        $functionName = $this->createTriggerFunctionName();
+
+        $this->connection->executeStatement(sprintf(
+            'DROP FUNCTION IF EXISTS %s() CASCADE;',
+            $functionName,
+        ));
+
+        $this->connection->executeStatement(sprintf(
+            'DROP TRIGGER IF EXISTS notify_trigger ON %s;',
+            $this->config['table_name'],
+        ));
+    }
+
     private function createTriggerFunctionName(): string
     {
         $tableConfig = explode('.', $this->config['table_name']);
diff --git a/src/Store/StreamDoctrineDbalStore.php b/src/Store/StreamDoctrineDbalStore.php
index e9fa93b40..ddef51f2b 100644
--- a/src/Store/StreamDoctrineDbalStore.php
+++ b/src/Store/StreamDoctrineDbalStore.php
@@ -442,6 +442,25 @@ public function setupSubscription(): void
         ));
     }
 
+    public function teardownSubscription(): void
+    {
+        if (!$this->supportSubscription()) {
+            return;
+        }
+
+        $functionName = $this->createTriggerFunctionName();
+
+        $this->connection->executeStatement(sprintf(
+            'DROP FUNCTION IF EXISTS %s() CASCADE;',
+            $functionName,
+        ));
+
+        $this->connection->executeStatement(sprintf(
+            'DROP TRIGGER IF EXISTS notify_trigger ON %s;',
+            $this->config['table_name'],
+        ));
+    }
+
     private function createTriggerFunctionName(): string
     {
         $tableConfig = explode('.', $this->config['table_name']);
diff --git a/src/Store/SubscriptionStore.php b/src/Store/SubscriptionStore.php
index b99ab0116..4256735eb 100644
--- a/src/Store/SubscriptionStore.php
+++ b/src/Store/SubscriptionStore.php
@@ -11,4 +11,6 @@ public function supportSubscription(): bool;
     public function setupSubscription(): void;
 
     public function wait(int $timeoutMilliseconds): void;
+
+    // public function teardownSubscription(): void;
 }