Add Twitch EventSub support with OAuth flow and command system

.gitignore

@@ -2,4 +2,5 @@
 /.idea
 composer.lock
 .phpunit.result.cache
+.twitch_token.json
 tmi.php

composer.json

@@ -22,6 +22,8 @@
   "require": {
     "php": "^7.4|^8.0",
     "react/socket": "^1.6",
+    "react/http": "^1.6",
+    "ratchet/pawl": "^0.4",
     "ext-mbstring": "*",
     "ext-json": "*"
   },

src/CommandContext.php

@@ -0,0 +1,43 @@
+<?php
+
+namespace GhostZero\Tmi;
+
+use GhostZero\Tmi\Events\EventSub\ChatMessageEvent;
+
+class CommandContext
+{
+    public ChatMessageEvent $event;
+    public string $command;
+
+    /**
+     * Arguments after the command name, split by spaces.
+     * @var string[]
+     */
+    public array $args;
+
+    private EventSubClient $client;
+
+    public function __construct(ChatMessageEvent $event, string $command, array $args, EventSubClient $client)
+    {
+        $this->event = $event;
+        $this->command = $command;
+        $this->args = $args;
+        $this->client = $client;
+    }
+
+    /**
+     * Sends a Twitch reply thread to the triggering message.
+     */
+    public function reply(string $message): void
+    {
+        $this->client->sendMessage($this->event->broadcasterUserId, $message, $this->event->messageId);
+    }
+
+    /**
+     * Sends a regular chat message (no reply thread).
+     */
+    public function say(string $message): void
+    {
+        $this->client->sendMessage($this->event->broadcasterUserId, $message);
+    }
+}

src/EventSubClient.php

@@ -0,0 +1,240 @@
+<?php
+
+namespace GhostZero\Tmi;
+
+use GhostZero\Tmi\Events\EventHandler;
+use GhostZero\Tmi\Events\EventSub\ChatMessageEvent;
+use GhostZero\Tmi\Events\EventSub\ChatNotificationEvent;
+use Ratchet\Client\WebSocket;
+use Ratchet\RFC6455\Messaging\MessageInterface;
+use React\EventLoop\LoopInterface;
+use React\Http\Browser;
+
+class EventSubClient
+{
+    private const WEBSOCKET_URL = 'wss://eventsub.wss.twitch.tv/ws';
+    private const API_SUBSCRIPTIONS_URL = 'https://api.twitch.tv/helix/eventsub/subscriptions';
+    private const API_CHAT_MESSAGES_URL = 'https://api.twitch.tv/helix/chat/messages';
+
+    private EventSubOptions $options;
+    private LoopInterface $loop;
+    private EventHandler $eventHandler;
+    private Browser $browser;
+    private array $commands = [];
+
+    public function __construct(EventSubOptions $options)
+    {
+        $this->options = $options;
+        $this->loop = \React\EventLoop\Factory::create();
+        $this->eventHandler = new EventHandler();
+        $this->browser = new Browser($this->loop);
+    }
+
+    public function connect(): void
+    {
+        $this->connectWebSocket(self::WEBSOCKET_URL);
+        $this->loop->run();
+    }
+
+    public function on(string $event, callable $closure): self
+    {
+        $this->eventHandler->addHandler($event, $closure);
+        return $this;
+    }
+
+    public function any(callable $closure): self
+    {
+        return $this->on('*', $closure);
+    }
+
+    /**
+     * Registers a chat command handler. The command name must include the prefix (e.g. '!hello').
+     * The handler receives a CommandContext with reply() and say() helpers.
+     */
+    public function command(string $name, callable $handler): self
+    {
+        $this->commands[strtolower($name)] = $handler;
+        return $this;
+    }
+
+    /**
+     * Sends a chat message via the Helix API. Pass a replyToMessageId to create a Twitch reply thread.
+     */
+    public function sendMessage(string $broadcasterUserId, string $message, ?string $replyToMessageId = null): void
+    {
+        $body = [
+            'broadcaster_id' => $broadcasterUserId,
+            'sender_id' => $this->options->getBotUserId(),
+            'message' => $message,
+        ];
+
+        if ($replyToMessageId !== null) {
+            $body['reply_parent_message_id'] = $replyToMessageId;
+        }
+
+        $this->browser->post(self::API_CHAT_MESSAGES_URL, [
+            'Authorization' => 'Bearer ' . $this->options->getAccessToken(),
+            'Client-Id' => $this->options->getClientId(),
+            'Content-Type' => 'application/json',
+        ], json_encode($body))->then(
+            function (\Psr\Http\Message\ResponseInterface $response) {
+                $this->debug("Chat message sent: HTTP {$response->getStatusCode()}");
+            },
+            function (\Exception $e) {
+                $this->debug("Failed to send chat message: " . $e->getMessage());
+            }
+        );
+    }
+
+    public function getLoop(): LoopInterface
+    {
+        return $this->loop;
+    }
+
+    public function getEventHandler(): EventHandler
+    {
+        return $this->eventHandler;
+    }
+
+    private function connectWebSocket(string $url): void
+    {
+        \Ratchet\Client\connect($url, [], [], $this->loop)->then(
+            function (WebSocket $conn) {
+                $this->debug("Connected to EventSub WebSocket.");
+
+                $conn->on('message', function (MessageInterface $msg) {
+                    $payload = json_decode((string) $msg, true);
+                    if ($payload !== null) {
+                        $this->handleMessage($payload);
+                    }
+                });
+
+                $conn->on('close', function (int $code, string $reason) {
+                    $this->debug("WebSocket closed: {$code} {$reason}");
+                    if ($this->options->shouldReconnect()) {
+                        $delay = $this->options->getReconnectDelay();
+                        $this->loop->addTimer($delay, fn () => $this->connectWebSocket(self::WEBSOCKET_URL));
+                    }
+                });
+
+                $conn->on('error', function (\Exception $e) {
+                    $this->debug("WebSocket error: " . $e->getMessage());
+                });
+            },
+            function (\Exception $e) {
+                $this->debug("Failed to connect: " . $e->getMessage());
+                if ($this->options->shouldReconnect()) {
+                    $delay = $this->options->getReconnectDelay();
+                    $this->loop->addTimer($delay, fn () => $this->connectWebSocket(self::WEBSOCKET_URL));
+                }
+            }
+        );
+    }
+
+    private function handleMessage(array $payload): void
+    {
+        $this->debug(json_encode($payload));
+
+        $messageType = $payload['metadata']['message_type'] ?? null;
+
+        switch ($messageType) {
+            case 'session_welcome':
+                $sessionId = $payload['payload']['session']['id'];
+                $this->subscribeToEvents($sessionId);
+                break;
+
+            case 'notification':
+                $this->handleNotification($payload['payload']);
+                break;
+
+            case 'session_reconnect':
+                $reconnectUrl = $payload['payload']['session']['reconnect_url'];
+                $this->debug("Reconnecting to: {$reconnectUrl}");
+                $this->connectWebSocket($reconnectUrl);
+                break;
+
+            case 'revocation':
+                $type = $payload['payload']['subscription']['type'] ?? 'unknown';
+                $status = $payload['payload']['subscription']['status'] ?? 'unknown';
+                $this->debug("Subscription revoked — type: {$type}, status: {$status}");
+                break;
+
+            case 'session_keepalive':
+                break;
+        }
+    }
+
+    private function subscribeToEvents(string $sessionId): void
+    {
+        foreach ($this->options->getSubscriptions() as $subscription) {
+            $body = json_encode([
+                'type' => $subscription['type'],
+                'version' => $subscription['version'] ?? '1',
+                'condition' => $subscription['condition'],
+                'transport' => [
+                    'method' => 'websocket',
+                    'session_id' => $sessionId,
+                ],
+            ]);
+
+            $this->browser->post(self::API_SUBSCRIPTIONS_URL, [
+                'Authorization' => 'Bearer ' . $this->options->getAccessToken(),
+                'Client-Id' => $this->options->getClientId(),
+                'Content-Type' => 'application/json',
+            ], $body)->then(
+                function (\Psr\Http\Message\ResponseInterface $response) use ($subscription) {
+                    $this->debug("Subscribed to {$subscription['type']}: HTTP {$response->getStatusCode()}");
+                },
+                function (\Exception $e) use ($subscription) {
+                    $this->debug("Failed to subscribe to {$subscription['type']}: " . $e->getMessage());
+                }
+            );
+        }
+    }
+
+    private function handleNotification(array $payload): void
+    {
+        $subscriptionType = $payload['subscription']['type'] ?? null;
+        $eventData = $payload['event'] ?? [];
+
+        $event = match ($subscriptionType) {
+            'channel.chat.message' => new ChatMessageEvent($eventData),
+            'channel.chat.notification' => new ChatNotificationEvent($eventData),
+            default => null,
+        };
+
+        if ($event !== null) {
+            $this->eventHandler->invoke($event);
+
+            if ($event instanceof ChatMessageEvent) {
+                $this->parseCommand($event);
+            }
+        }
+    }
+
+    private function parseCommand(ChatMessageEvent $event): void
+    {
+        $text = trim($event->messageText);
+
+        if ($text === '' || $text[0] !== '!') {
+            return;
+        }
+
+        $parts = explode(' ', $text, 2);
+        $name = strtolower($parts[0]);
+        $args = isset($parts[1]) ? explode(' ', trim($parts[1])) : [];
+
+        if (!isset($this->commands[$name])) {
+            return;
+        }
+
+        ($this->commands[$name])(new CommandContext($event, $name, $args, $this));
+    }
+
+    private function debug(string $message): void
+    {
+        if ($this->options->isDebug()) {
+            print $message . PHP_EOL;
+        }
+    }
+}

src/EventSubOptions.php

@@ -0,0 +1,73 @@
+<?php
+
+namespace GhostZero\Tmi;
+
+class EventSubOptions
+{
+    private array $options;
+
+    public function __construct(array $options)
+    {
+        $this->options = $options;
+    }
+
+    public function getAccessToken(): string
+    {
+        return ltrim($this->options['access_token'], 'oauth:');
+    }
+
+    public function getClientId(): string
+    {
+        return $this->options['client_id'];
+    }
+
+    /**
+     * Returns the list of EventSub subscriptions to register.
+     *
+     * Each entry must have:
+     *   - type:      e.g. 'channel.chat.message'
+     *   - version:   (optional, defaults to '1')
+     *   - condition: associative array of condition fields
+     *
+     * Example:
+     *   ['type' => 'channel.chat.message', 'condition' => ['broadcaster_user_id' => '123', 'user_id' => '456']]
+     */
+    public function getSubscriptions(): array
+    {
+        return $this->options['subscriptions'] ?? [];
+    }
+
+    /**
+     * Returns the bot's Twitch user ID used as the sender when posting chat messages.
+     * Reads from the top-level 'user_id' option, falling back to the first subscription's condition.
+     */
+    public function getBotUserId(): string
+    {
+        if (isset($this->options['user_id'])) {
+            return $this->options['user_id'];
+        }
+
+        foreach ($this->options['subscriptions'] ?? [] as $sub) {
+            if (isset($sub['condition']['user_id'])) {
+                return $sub['condition']['user_id'];
+            }
+        }
+
+        throw new \RuntimeException('Bot user ID not configured. Add "user_id" to EventSubOptions or include it in a subscription condition.');
+    }
+
+    public function isDebug(): bool
+    {
+        return $this->options['options']['debug'] ?? false;
+    }
+
+    public function shouldReconnect(): bool
+    {
+        return $this->options['connection']['reconnect'] ?? true;
+    }
+
+    public function getReconnectDelay(): int
+    {
+        return $this->options['connection']['reconnect_delay'] ?? 3;
+    }
+}