Feature: Orchestration

fakes/LoopFake.php

@@ -0,0 +1,89 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Fakes\Ragnarok\Fenrir;
+
+use Ragnarok\Fenrir\DataMapper;
+use Psr\Log\NullLogger;
+use React\EventLoop\LoopInterface;
+use React\EventLoop\TimerInterface;
+
+class LoopFake implements LoopInterface
+{
+    private array $timers = [];
+
+    public function addReadStream($stream, $listener)
+    {
+    }
+
+    public function addWriteStream($stream, $listener)
+    {
+    }
+
+    public function removeReadStream($stream)
+    {
+    }
+
+    public function removeWriteStream($stream)
+    {
+    }
+
+    public function addTimer($interval, $callback)
+    {
+        $this->timers[] = ['seconds' => $interval, 'callback' => $callback];
+    }
+
+    public function runTimers(?int $seconds = null)
+    {
+        $timers = $this->timers;
+        $this->timers = [];
+
+        if ($seconds === null) {
+            foreach ($timers as $i => $timer) {
+                $timer['callback']();
+            }
+
+            return;
+        }
+
+        foreach ($timers as $i => &$timer) {
+            $timer['seconds'] -= $seconds;
+
+            if ($timer['seconds'] <= 0) {
+                $timer['callback']();
+                unset($timers[$i]);
+            }
+        }
+
+        $this->timers = [...$timers, ...$this->timers];
+    }
+
+    public function addPeriodicTimer($interval, $callback)
+    {
+    }
+
+    public function cancelTimer(TimerInterface $timer)
+    {
+    }
+
+    public function futureTick($listener)
+    {
+    }
+
+    public function addSignal($signal, $listener)
+    {
+    }
+
+    public function removeSignal($signal, $listener)
+    {
+    }
+
+    public function run()
+    {
+    }
+
+    public function stop()
+    {
+    }
+}

src/Orchestration/Orchestrator.php

@@ -0,0 +1,137 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Ragnarok\Fenrir\Orchestration;
+
+use Carbon\Carbon;
+use DateTimeInterface;
+use Evenement\EventEmitter;
+use Ragnarok\Fenrir\Parts\GatewayBot;
+use Ragnarok\Fenrir\Parts\SessionStartLimit;
+use Ragnarok\Fenrir\Rest\Gateway;
+use React\EventLoop\LoopInterface;
+use React\Promise\Promise;
+use React\Promise\PromiseInterface;
+
+class Orchestrator extends EventEmitter
+{
+    public const ALLOW_SPAWN = 'allow_spawn';
+
+    public const CONCURRENCY_TIMESPAN_SECONDS = 5;
+
+    /**
+     * @var int[]
+     */
+    private array $shards;
+
+    private int $totalShards;
+
+    public function __construct(
+        private readonly LoopInterface $loop,
+        private readonly Gateway $gateway,
+        private readonly ?int $totalShardsOverride = null
+    ) {
+    }
+
+    /**
+     * @return PromiseInterface<null>
+     */
+    public function init(): PromiseInterface
+    {
+        return $this->gateway->getBot()->then(function (GatewayBot $bot) {
+            $this->totalShards = $this->totalShardsOverride ?? $bot->shards;
+
+            $this->shards = range(
+                0,
+                $this->totalShards - 1
+            );
+
+            return $this->orchestrate($bot->session_start_limit);
+        });
+    }
+
+    /**
+     * @return PromiseInterface<null>
+     */
+    private function orchestrate(SessionStartLimit $sessionStartLimits): PromiseInterface
+    {
+        return $this->startSpawning(
+            $sessionStartLimits->remaining,
+            $sessionStartLimits->max_concurrency,
+            Carbon::createFromTimestamp($sessionStartLimits->reset_after),
+        )->then(function () use ($sessionStartLimits) {
+            $wait = max(
+                $sessionStartLimits->reset_after - (Carbon::now())->getTimestamp(),
+                1
+            );
+
+            return new Promise(function (callable $resolve) use ($wait) {
+                $this->loop->addTimer(
+                    $wait,
+                    fn () => $this->continueOrchestration()->then(fn () => $resolve(null))
+                );
+            });
+        });
+    }
+
+    /**
+     * @return PromiseInterface<null>
+     */
+    private function continueOrchestration(): PromiseInterface
+    {
+        return new Promise(function (callable $resolve) {
+            if (!empty($this->shards)) {
+                return $this->gateway->getBot()->then(
+                    fn (GatewayBot $bot) => $this->orchestrate($bot->session_start_limit)
+                        ->then(fn () => $resolve(null))
+                );
+            }
+        });
+    }
+
+    /**
+     * @return PromiseInterface<null>
+     */
+    private function startSpawning(int $remaining, int $concurrency, DateTimeInterface $resetAfter): PromiseInterface
+    {
+        return new Promise(function (callable $resolve) use ($remaining, $concurrency, $resetAfter) {
+            $currentDate = Carbon::now();
+
+            if ($currentDate > $resetAfter) {
+                $resolve(null);
+
+                return;
+            }
+
+            if ($remaining === 0) {
+                $resolve(null);
+                return;
+            }
+
+            $shardsToSpawn = $remaining > $concurrency
+                ? $concurrency
+                : $remaining;
+
+            $remaining = max($remaining - $concurrency, 0);
+
+            $shards = array_splice($this->shards, 0, $shardsToSpawn);
+
+            $this->allowSpawning($shards);
+
+            $this->loop->addTimer(self::CONCURRENCY_TIMESPAN_SECONDS, function () use ($remaining, $concurrency, $resetAfter, $resolve) {
+                $this->startSpawning($remaining, $concurrency, $resetAfter)->then(fn () => $resolve(null));
+            });
+        });
+    }
+
+    /**
+     * @param int[] $shards
+     */
+    private function allowSpawning(array $shards): void
+    {
+        foreach ($shards as $shard) {
+            $this->emit(self::ALLOW_SPAWN, [$shard, $this->totalShards]);
+        }
+    }
+}

src/Parts/Gateway.php

@@ -0,0 +1,10 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Ragnarok\Fenrir\Parts;
+
+class Gateway
+{
+    public string $url;
+}

src/Parts/GatewayBot.php

@@ -0,0 +1,12 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Ragnarok\Fenrir\Parts;
+
+class GatewayBot
+{
+    public string $url;
+    public int $shards;
+    public SessionStartLimit $session_start_limit;
+}

src/Parts/SessionStartLimit.php

@@ -0,0 +1,13 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Ragnarok\Fenrir\Parts;
+
+class SessionStartLimit
+{
+    public int $total;
+    public int $remaining;
+    public int $reset_after;
+    public int $max_concurrency;
+}

src/Rest/Gateway.php

@@ -0,0 +1,36 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Ragnarok\Fenrir\Rest;
+
+use Discord\Http\Endpoint;
+use Ragnarok\Fenrir\Parts\Gateway as PartsGateway;
+use Ragnarok\Fenrir\Parts\GatewayBot;
+use React\Promise\PromiseInterface;
+
+/**
+ * @see https://discord.com/developers/docs/events/gateway
+ */
+class Gateway extends HttpResource
+{
+    public function get(): PromiseInterface
+    {
+        return $this->mapPromise(
+            $this->http->get(
+                Endpoint::GATEWAY
+            ),
+            PartsGateway::class,
+        );
+    }
+
+    public function getBot(): PromiseInterface
+    {
+        return $this->mapPromise(
+            $this->http->get(
+                Endpoint::GATEWAY_BOT
+            ),
+            GatewayBot::class,
+        );
+    }
+}

src/Rest/Rest.php

@@ -30,6 +30,7 @@ class Rest
     public readonly GlobalCommand $globalCommand;
     public readonly Webhook $webhook;
     public readonly Guild $guild;
+    public readonly Gateway $gateway;
 
     public function __construct(private Http $http, private DataMapper $dataMapper, private LoggerInterface $logger)
     {
@@ -52,5 +53,6 @@ public function __construct(private Http $http, private DataMapper $dataMapper,
         $this->globalCommand = new GlobalCommand(...$args);
         $this->webhook = new Webhook(...$args);
         $this->guild = new Guild(...$args);
+        $this->gateway = new Gateway(...$args);
     }
 }