README.md

@@ -2,52 +2,4 @@
 
 [GitLab](https://gitlab.tugraz.at/dbp/relay/dbp-relay-core-bundle) | [Packagist](https://packagist.org/packages/dbp/relay-core-bundle)
 
-## Bundle Config
-
-Created via `./bin/console config:dump-reference DbpRelayCoreBundle | sed '/^$/d'`
-
-```yaml
-# Default configuration for "DbpRelayCoreBundle"
-dbp_relay_core:
-    # Some string identifying the current build (commit hash)
-    build_info:              ~ # Example: deadbeef
-    # Some URL identifying the current build (URL to the commit on some git web interface)
-    build_info_url:          ~ # Example: 'https://gitlab.example.com/project/-/commit/deadbeef'
-    # The title text of the API docs page
-    docs_title:              'Relay API Gateway'
-    # The description text of the API docs page (supports markdown)
-    docs_description:        '*part of the [Digital Blueprint](https://gitlab.tugraz.at/dbp) project*'
-    lock_dsn:                ''
-    messenger_transport_dsn: '%env(MESSENGER_TRANSPORT_DSN)'
-```
-
-### Locking
-
-To handle [locking](https://symfony.com/doc/current/components/lock.html) you need to set above `lock_dsn` config,
-for example as `lock_dsn: '%env(LOCK_DSN)%'` with an environment variable `LOCK_DSN` in your `.env` file or by any other means.
-
-For example, you could use [Redis](https://redis.io/) for distributed locking or `semaphore` for local locking.
-
-Example:
-
-```dotenv
-# Redis (distributed locking)
-LOCK_DSN=redis://redis:6379/
-
-# Semaphore (local locking)
-LOCK_DSN=semaphore
-```
-
-### Symfony Messenger
-
-For projects that also use the [Symfony Messenger](https://symfony.com/doc/current/components/messenger.html)
-you need to set above `messenger_transport_dsn` config, for example as `messenger_transport_dsn: '%env(MESSENGER_TRANSPORT_DSN)%'`
-with an environment variable `MESSENGER_TRANSPORT_DSN` in your `.env` file or by any other means.
-
-[Redis](https://redis.io/) is also a way for doing this.
-
-Example:
-
-```dotenv
-MESSENGER_TRANSPORT_DSN=redis://redis:6379/local-messages/symfony/consumer?auto_setup=true&serializer=1&stream_max_entries=0&dbindex=0
-```
+Docs: see <https://gitlab.tugraz.at/dbp/relay/dbp-relay-core-bundle/-/tree/main/docs>

composer.json

@@ -17,6 +17,7 @@
         "symfony/event-dispatcher": "^5.3",
         "symfony/expression-language": "^5.3",
         "symfony/framework-bundle": "^5.3",
+        "symfony/messenger": "^5.3",
         "symfony/mime": "^5.3",
         "symfony/security-bundle": "^5.3",
         "symfony/security-core": "^5.3",

docs/config.md

+# Bundle Configuration
+
+Created via `./bin/console config:dump-reference DbpRelayCoreBundle | sed '/^$/d'`
+
+```yaml
+# Default configuration for "DbpRelayCoreBundle"
+dbp_relay_core:
+    # Some string identifying the current build (commit hash)
+    build_info:           ~ # Example: deadbeef
+    # Some URL identifying the current build (URL to the commit on some git web interface)
+    build_info_url:       ~ # Example: 'https://gitlab.example.com/project/-/commit/deadbeef'
+    # The title text of the API docs page
+    docs_title:           'Relay API Gateway'
+    # The description text of the API docs page (supports markdown)
+    docs_description:     '*part of the [Digital Blueprint](https://gitlab.tugraz.at/dbp) project*'
+    messenger_transport_dsn: '' # Deprecated (Since dbp/relay-core-bundle 0.1.20: Use "queue_dsn" instead.)
+    # See https://symfony.com/doc/5.3/messenger.html#redis-transport
+    queue_dsn:            '' # Example: 'redis://localhost:6379'
+    # https://symfony.com/doc/5.3/components/lock.html
+    lock_dsn:             '' # Example: 'redis://redis:6379'
+```
+
+## Locking
+
+To handle [locking](https://symfony.com/doc/current/components/lock.html) you need to set above `lock_dsn` config,
+for example as `lock_dsn: '%env(LOCK_DSN)%'` with an environment variable `LOCK_DSN` in your `.env` file or by any other means.
+
+For example, you could use [Redis](https://redis.io/) for distributed locking or `semaphore` for local locking.
+
+Example:
+
+```dotenv
+# Redis (distributed locking)
+LOCK_DSN=redis://redis:6379/
+
+# Semaphore (local locking)
+LOCK_DSN=semaphore
+```

docs/cron.md

+# Cron Jobs
+
+The API gateway provides one shared cron command which you should call every few
+minutes:
+
+```bash
+./bin/console dbp:relay:core:cron
+```
+
+For example in crontab, every 5 minutes:
+
+```bash
+*/5 * * * * /srv/api/bin/console dbp:relay:core:cron
+```
+
+This cron job will regularly prune caches and dispatch a cron event which can be
+handled by different bundles.
+
+To get access to such an even you have to implement an event listener:
+
+```yaml
+  Dbp\Relay\MyBundle\Cron\CleanupJob:
+    tags:
+      - { name: kernel.event_listener, event: dbp.relay.cron }
+```
+
+The listener gets called with a `CronEvent` object. By calling
+`CronEvent::isDue()` and passing an ID for logging and a  [cron
+expression](https://en.wikipedia.org/wiki/Cron) you get told when it is time to
+run:
+
+```php
+class CleanupJob
+{
+    public function onDbpRelayCron(CronEvent $event)
+    {
+        if ($event->isDue('mybundle-cleanup', '0 * * * *')) {
+            // Do cleanup things here..
+        }
+    }
+}
+```

docs/errors.md

+# API Errors and Error Handling
+
+By default Symfony and API Platform convert `HttpException` and all subclasses
+to HTTP errors with a matching status code. See
+https://api-platform.com/docs/core/errors for details.
+
+Since API Platform by default hides any message details for >= 500 and < 600 in
+production and doesn't allow injecting any extra information into the resulting
+JSON-LD error response we provide a special HttpException subclass which
+provides those features.
+
+The following will pass the error message to the client even in case the status
+code is 5xx. Note that you have to be careful to not include any secrets in the
+error message since they would be exposed to the client.
+
+```php
+use Dbp\Relay\CoreBundle\Exception\ApiError;
+
+throw new APIError(500, 'My custom message');
+```
+
+which results in:
+
+```json
+{
+  "@context": "/contexts/Error",
+  "@type": "hydra:Error",
+  "hydra:title": "An error occurred",
+  "hydra:description": "My custom message"
+}
+```
+
+Further more you can include extra information like an error ID and some extra
+information in form of an object:
+
+```php
+use Dbp\Relay\CoreBundle\Exception\ApiError;
+
+throw new APIError::withDetails(500, 'My custom message', 'my-id', ['foo' => 42]);
+```
+
+which results in:
+
+```json
+{
+  "@context": "/contexts/Error",
+  "@type": "hydra:Error",
+  "hydra:title": "An error occurred",
+  "hydra:description": "My custom message",
+  "relay:errorId": "my-id",
+  "relay:errorDetails": {
+    "foo": 42
+  }
+```
+
+If you are using status codes <= 400 and are fine with just the message, then
+using any of the builtin exception types is fine.

docs/index.md

+# About
+
+The core bundle is the central bundle that needs to be installed in every API
+gateway and also is a dependency of every other API bundle.
+
+* It provides functionality that is commonly needed by API bundles (error handling,
+  logging, etc)
+* It integrates the auth bundle with the Symfony security system
+* It provides console commands that API bundles can subscribe to
+* It configures all dependencies to our needs (api-platform, symfony, etc.)
+* and more ...
+
+
+# Overview
+
+A minimal working relay API gateway consists of the core bundle and an auth bundle.
+
+```mermaid
+graph TD
+    style core_bundle fill:#606096,color:#fff
+
+    subgraph API Gateway
+        api(("API"))
+        auth_bundle("Auth Bundle")
+        core_bundle("Core Bundle")
+        core_bundle --> auth_bundle
+        api --> core_bundle
+    end
+
+    click auth_bundle "./#auth-bundle"
+    click user_session "./#usersession"
+```
+
+### Auth Bundle
+
+The auth bundle takes care of user authentication and communicates with an OIDC
+server, for example [Keycloak](https://www.keycloak.org). It creates the Symfony
+user object and converts OAuth2 scopes to Symfony user roles used for
+authorization.

docs/queue.md

+# Queued Tasks
+
+The Relay API gateway optionally requires a queuing system, which means tasks
+get queued in a central data store and worked on after a request has finished.
+The tasks can be processes using one or more workers on multiple machines in
+parallel.
+
+This requires two extra deployment related tasks:
+
+1) One or more worker tasks have to be run in the background and automatically
+   restarted if they stop.
+2) On deployment the worker processes have to be restarted to use the new code.
+
+## Configuration
+
+In the bundle configuration set the `queue_dsn` key to a DSN supported by the
+[Symfony messenger component](https://symfony.com/doc/current/messenger.html)
+
+At the moment we only support the redis transport.
+
+Example:
+
+```yaml
+queue_dsn: 'redis://localhost:6379'
+```
+
+## Run the workers
+
+Start a worker using
+
+```bash
+./bin/console dbp:relay:core:queue:work my-worker-01
+```
+
+It will automatically exit after a specific amount 0f time or after a specific
+number of processed tasks.
+
+Note:
+
+* You need to take care of restarting it automatically.
+* Each active worker needs to have a unique name passed as the first argument
+  which should stay the same across restarts.
+
+
+## Restart the workers
+
+After deployment run
+
+```bash
+./bin/console dbp:relay:core:queue:restart
+```
+
+This will signal the workers to exit after the current task, which means they
+will be restarted by supervisor and will run the newly deployed code.
+
+Symfony
+[recommends](https://symfony.com/doc/current/messenger.html#supervisor-configuration)
+to use [Supervisor](http://supervisord.org/) to do this.
+
+```bash
+sudo apt-get install supervisor
+```
+
+```ini
+;/etc/supervisor/conf.d/queue-worker.conf
+[program:queue-work]
+command=php /path/to/your/app/bin/console dbp:relay:core:queue:work "%(program_name)s_%(process_num)02d"
+user=user
+numprocs=2
+startsecs=0
+autostart=true
+autorestart=true
+process_name=%(program_name)s_%(process_num)02d
+```
+
+Change `user` to the Unix user on your server.
\ No newline at end of file

src/Cron/CronCommand.php

@@ -18,7 +18,8 @@ final class CronCommand extends Command implements LoggerAwareInterface
 {
     use LoggerAwareTrait;
 
-    protected static $defaultName = 'dbp:relay:cron|dbp:cron';
+    // dbp:cron only for backwards compat
+    protected static $defaultName = 'dbp:relay:core:cron|dbp:cron';
 
     /** @var EventDispatcherInterface */
     private $dispatcher;

src/DependencyInjection/Configuration.php

@@ -31,9 +31,13 @@ class Configuration implements ConfigurationInterface
                     ->defaultValue('*part of the [Digital Blueprint](https://gitlab.tugraz.at/dbp) project*')
                 ->end()
                 ->scalarNode('messenger_transport_dsn')
+                    ->defaultValue('')
+                    ->setDeprecated('dbp/relay-core-bundle', '0.1.20', 'Use "queue_dsn" instead.')
+                ->end()
+                ->scalarNode('queue_dsn')
                     ->info('See https://symfony.com/doc/5.3/messenger.html#redis-transport')
                     ->defaultValue('')
-                    ->example('redis://localhost:6379/messages')
+                    ->example('redis://redis:6379')
                 ->end()
                 ->scalarNode('lock_dsn')
                     ->info('https://symfony.com/doc/5.3/components/lock.html')

src/DependencyInjection/DbpRelayCoreExtension.php

@@ -164,7 +164,11 @@ class DbpRelayCoreExtension extends ConfigurableExtension implements PrependExte
         ]);
 
         // https://symfony.com/doc/4.4/messenger.html#transports-async-queued-messages
-        $messengerTransportDsn = $config['messenger_transport_dsn'];
+        $messengerTransportDsn = $config['queue_dsn'];
+        if ($messengerTransportDsn === '') {
+            // backward compatibility
+            $messengerTransportDsn = $config['messenger_transport_dsn'];
+        }
         if ($container->hasParameter('dbp_api.messenger_routing')) {
             $routing = [];
             $routing = array_merge($routing, $container->getParameter('dbp_api.messenger_routing'));

src/Queue/RestartCommand.php

+<?php
+
+declare(strict_types=1);
+
+namespace Dbp\Relay\CoreBundle\Queue;
+
+use Psr\Log\LoggerAwareInterface;
+use Psr\Log\LoggerAwareTrait;
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Input\ArrayInput;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+
+class RestartCommand extends Command implements LoggerAwareInterface
+{
+    use LoggerAwareTrait;
+
+    protected static $defaultName = 'dbp:relay:core:queue:restart';
+
+    protected function configure()
+    {
+        $this->setDescription('Stop all workers once they are done with the current task');
+    }
+
+    protected function execute(InputInterface $input, OutputInterface $output)
+    {
+        // Now run the real messenger:stop-workers command
+        $app = $this->getApplication();
+        assert($app !== null);
+        $command = $app->find('messenger:stop-workers');
+        $consumeInput = new ArrayInput([]);
+
+        return $command->run($consumeInput, $output);
+    }
+}

src/Queue/TransportFactoryDecorator.php

+<?php
+
+declare(strict_types=1);
+
+namespace Dbp\Relay\CoreBundle\Queue;
+
+use Psr\Log\LoggerAwareInterface;
+use Psr\Log\LoggerAwareTrait;
+use Symfony\Component\Messenger\Bridge\Redis\Transport\RedisTransportFactory;
+use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
+use Symfony\Component\Messenger\Transport\TransportFactoryInterface;
+use Symfony\Component\Messenger\Transport\TransportInterface;
+
+/**
+ * Decorated because we want to set reasonable defaults while still allowing the user to set a full transport
+ * DSN if needed.
+ */
+class TransportFactoryDecorator implements TransportFactoryInterface, LoggerAwareInterface
+{
+    use LoggerAwareTrait;
+
+    /**
+     * @var TransportFactoryInterface
+     */
+    private $decorated;
+
+    /**
+     * @var string
+     */
+    private $workerName;
+
+    public function __construct(TransportFactoryInterface $decorated)
+    {
+        $this->decorated = $decorated;
+        $this->workerName = 'worker';
+    }
+
+    public function setActiveWorkerName(string $name)
+    {
+        $this->workerName = $name;
+    }
+
+    public function createTransport(string $dsn, array $options, SerializerInterface $serializer): TransportInterface
+    {
+        // In case we build our main transport we make sure the defaults are set to our liking.
+        // The DSN content still wins, but ideally the user only sets the minimal DSN.
+        if ($options['transport_name'] === Utils::QUEUE_TRANSPORT_NAME) {
+            $this->logger->debug("Creating queue transport for worker: '$this->workerName'");
+
+            $redis = new RedisTransportFactory();
+            if ($redis->supports($dsn, $options)) {
+                // We set some nice namespaced default, so the user doesn't have to care about potential conflicts
+                $options['stream'] = 'dbp_relay_queue_stream';
+                $options['group'] = 'dbp_relay_queue_group';
+                $options['consumer'] = $this->workerName;
+                // Use the new recommended default:
+                // https://github.com/symfony/symfony/pull/42163
+                $options['delete_after_ack'] = true;
+            } else {
+                throw new \Exception('Only redis currently supported as a messenger transport (current DSN: '.$dsn.')');
+            }
+        }
+
+        return $this->decorated->createTransport($dsn, $options, $serializer);
+    }
+
+    public function supports(string $dsn, array $options): bool
+    {
+        return $this->decorated->supports($dsn, $options);
+    }
+}

src/Queue/Utils.php

+<?php
+
+declare(strict_types=1);
+
+namespace Dbp\Relay\CoreBundle\Queue;
+
+class Utils
+{
+    public const QUEUE_TRANSPORT_NAME = 'async';
+
+    // These just should be good defaults, feel free to adjust
+    public const DEFAULT_TIME_LIMIT = 3600;
+    public const DEFAULT_TASK_LIMIT = 10;
+}

src/Queue/WorkCommand.php

+<?php
+
+declare(strict_types=1);
+
+namespace Dbp\Relay\CoreBundle\Queue;
+
+use Psr\Log\LoggerAwareInterface;
+use Psr\Log\LoggerAwareTrait;
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Input\ArrayInput;
+use Symfony\Component\Console\Input\InputArgument;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+
+/**
+ * This is a simpler variant of "messenger:consume". It works with a fixed transport and by requiring a unique
+ * worker name works the same across redis and DB transports.
+ *
+ * It also sets some nice defaults for everything.
+ */
+class WorkCommand extends Command implements LoggerAwareInterface
+{
+    use LoggerAwareTrait;
+
+    protected static $defaultName = 'dbp:relay:core:queue:work';
+
+    /**
+     * @var TransportFactoryDecorator
+     */
+    private $transportFactory;
+
+    public function __construct(TransportFactoryDecorator $transportFactory)
+    {
+        parent::__construct();
+        $this->transportFactory = $transportFactory;
+    }
+
+    protected function configure(): void
+    {
+        $this->setDescription('Start a worker which processes queued tasks');
+        $this->addArgument('worker-name', InputArgument::REQUIRED, 'A unique and stable worker name');
+    }
+
+    protected function execute(InputInterface $input, OutputInterface $output)
+    {
+        // The question is why can't we use a random worker name based on the hostname or pid etc.
+        // https://github.com/symfony/symfony-docs/pull/11869/files explains why consumer names
+        // should be unique, reused, and ideally stable, when using redis. By requiring this for all transports
+        // we keep the config/docs simple.
+        $workerName = $input->getArgument('worker-name');
+        $this->transportFactory->setActiveWorkerName($workerName);
+
+        // Now run the real messenger:consume command
+        $app = $this->getApplication();
+        assert($app !== null);
+        $command = $app->find('messenger:consume');
+        // use some good default limits, since lots of php code leaks the workers have to be restarted from time to time
+        $consumeInput = new ArrayInput([
+            '--time-limit' => (string) Utils::DEFAULT_TIME_LIMIT,
+            '--limit' => (string) Utils::DEFAULT_TASK_LIMIT,
+            'receivers' => [Utils::QUEUE_TRANSPORT_NAME],
+        ]);
+
+        return $command->run($consumeInput, $output);
+    }
+}

src/Resources/config/services.yaml

@@ -18,3 +18,16 @@ services:
     decorates: 'api_platform.openapi.factory'
     autowire: true
     autoconfigure: false
+
+  Dbp\Relay\CoreBundle\Queue\TransportFactoryDecorator:
+    decorates: 'messenger.transport_factory'
+    autowire: true
+    autoconfigure: true
+
+  Dbp\Relay\CoreBundle\Queue\WorkCommand:
+    autowire: true
+    autoconfigure: true
+
+  Dbp\Relay\CoreBundle\Queue\RestartCommand:
+    autowire: true
+    autoconfigure: true
\ No newline at end of file

tests/Queue/RestartCommandTest.php

+<?php
+
+declare(strict_types=1);
+
+namespace Dbp\Relay\CoreBundle\Tests\Queue;
+
+use Symfony\Bundle\FrameworkBundle\Console\Application;
+use Symfony\Bundle\FrameworkBundle\Test\KernelTestCase;
+use Symfony\Component\Console\Tester\CommandTester;
+
+class RestartCommandTest extends KernelTestCase
+{
+    public function testExecute()
+    {
+        $kernel = static::createKernel();
+        $application = new Application($kernel);
+
+        $command = $application->find('dbp:relay:queue:restart');
+        $commandTester = new CommandTester($command);
+        $res = $commandTester->execute([]);
+        $this->assertSame(0, $res);
+    }
+}