From 140dd4b08e7346c915df60905c4f12aba2b19cb2 Mon Sep 17 00:00:00 2001 From: Logan Lindquist Land Date: Wed, 16 Sep 2026 08:52:18 -0500 Subject: [PATCH 1/2] feat: add private messaging channels --- README.md | 1 + app/Config/Routes.php | 9 + app/Config/Services.php | 15 +- app/Contracts/ChannelRepository.php | 38 ++ app/Contracts/ChatRepository.php | 5 +- app/Controllers/Api/V1/ChannelsController.php | 367 ++++++++++++++++++ .../Api/V1/ReactionsController.php | 32 +- .../2026-09-16-000000_create_channels.php | 115 ++++++ app/Libraries/ChatWebSocketServer.php | 178 ++++++++- app/Models/ChannelModel.php | 210 ++++++++++ app/Models/ChatModel.php | 49 ++- app/Views/chat/svelteView.php | 2 + app/Views/chat/vueView.php | 2 + docs/api-versioning.md | 4 + docs/channels.md | 109 ++++++ e2e/support/chat-scenarios.ts | 6 + src/svelte/App.svelte | 282 +++++++++++++- src/svelte/main.js | 2 + src/vue/App.vue | 269 ++++++++++++- src/vue/main.js | 2 + .../Integration/ChannelApiIntegrationTest.php | 181 +++++++++ .../ChannelModelIntegrationTest.php | 107 +++++ .../WebSocketFlowIntegrationTest.php | 99 ++++- tests/Tests/Unit/ChatModelTest.php | 8 +- tests/Tests/Unit/ChatWebSocketTypingTest.php | 23 +- tests/Tests/Unit/DependencyInjectionTest.php | 2 + tests/_support/IntegrationTestCase.php | 2 +- 27 files changed, 2024 insertions(+), 95 deletions(-) create mode 100644 app/Contracts/ChannelRepository.php create mode 100644 app/Controllers/Api/V1/ChannelsController.php create mode 100644 app/Database/Migrations/2026-09-16-000000_create_channels.php create mode 100644 app/Models/ChannelModel.php create mode 100644 docs/channels.md create mode 100644 tests/Tests/Integration/ChannelApiIntegrationTest.php create mode 100644 tests/Tests/Integration/ChannelModelIntegrationTest.php diff --git a/README.md b/README.md index 77193f6..d0429fc 100644 --- a/README.md +++ b/README.md @@ -161,6 +161,7 @@ See: - `docs/public-api.md` (HTTP + WebSocket + frontend entrypoints + reusable PHP APIs) - `docs/api.md` (message reaction HTTP and WebSocket contracts) - `docs/user-profiles.md` (profile settings, avatars, themes, and WebSocket presence) +- `docs/channels.md` (public channels, private DMs, APIs, WebSocket routing, and migration operations) ## Troubleshooting diff --git a/app/Config/Routes.php b/app/Config/Routes.php index c8e83df..3805f18 100644 --- a/app/Config/Routes.php +++ b/app/Config/Routes.php @@ -35,6 +35,15 @@ $routes->get('messages', 'MessagesController::list', ['filter' => ['apiFormat', 'auth']]); $routes->get('messages/search', 'MessagesController::search', ['filter' => ['apiFormat', 'auth']]); $routes->post('messages', 'MessagesController::create', ['filter' => ['apiFormat', 'auth', 'rate:write', 'validate:message']]); + $routes->get('channels', 'ChannelsController::list', ['filter' => ['apiFormat', 'auth']]); + $routes->post('channels', 'ChannelsController::create', ['filter' => ['apiFormat', 'auth', 'rate:write']]); + $routes->patch('channels/(:num)', 'ChannelsController::update/$1', ['filter' => ['apiFormat', 'auth', 'rate:write']]); + $routes->post('channels/(:num)/join', 'ChannelsController::join/$1', ['filter' => ['apiFormat', 'auth', 'rate:write']]); + $routes->post('channels/(:num)/leave', 'ChannelsController::leave/$1', ['filter' => ['apiFormat', 'auth', 'rate:write']]); + $routes->get('channels/(:num)/messages', 'ChannelsController::messages/$1', ['filter' => ['apiFormat', 'auth']]); + $routes->get('channels/(:num)/messages/search', 'ChannelsController::searchMessages/$1', ['filter' => ['apiFormat', 'auth']]); + $routes->post('channels/(:num)/messages', 'ChannelsController::createMessage/$1', ['filter' => ['apiFormat', 'auth', 'rate:write']]); + $routes->post('dms', 'ChannelsController::createDm', ['filter' => ['apiFormat', 'auth', 'rate:write']]); $routes->get('messages/(:num)/reactions', 'ReactionsController::list/$1', ['filter' => ['apiFormat', 'auth']]); $routes->post('messages/(:num)/reactions', 'ReactionsController::create/$1', ['filter' => ['apiFormat', 'auth', 'rate:react']]); $routes->delete('messages/(:num)/reactions/(:segment)', 'ReactionsController::delete/$1/$2', ['filter' => ['apiFormat', 'auth', 'rate:react']]); diff --git a/app/Config/Services.php b/app/Config/Services.php index 8cb7232..5ec37ba 100644 --- a/app/Config/Services.php +++ b/app/Config/Services.php @@ -4,12 +4,14 @@ use App\Contracts\AuditLogger as AuditLoggerContract; use App\Contracts\AuditLogRepository; +use App\Contracts\ChannelRepository; use App\Contracts\ChatFormatter as ChatFormatterContract; use App\Contracts\ChatRepository; use App\Contracts\CspReportRepository; use App\Contracts\PushSubscriptionRepository; use App\Contracts\ReactionRepository; use App\Contracts\UserRepository; +use App\Controllers\Api\V1\ChannelsController; use App\Controllers\Api\V1\MessagesController; use App\Controllers\Api\V1\PushSubscriptionsController; use App\Controllers\Api\V1\ReactionsController; @@ -22,6 +24,7 @@ use App\Core\Application; use App\Libraries\ErrorHandler; use App\Models\AuditLogModel; +use App\Models\ChannelModel; use App\Models\ChatModel; use App\Models\CspReportModel; use App\Models\MessageReactionModel; @@ -71,8 +74,9 @@ public static function controller(string $controllerClass): Controller Home::class => new Home(), Chat::class => new Chat(static::chatRepository(), static::chatFormatter()), MessagesController::class => new MessagesController(static::chatRepository(), static::chatFormatter()), + ChannelsController::class => new ChannelsController(static::channelRepository(), static::chatRepository(), static::userRepository()), PushSubscriptionsController::class => new PushSubscriptionsController(static::pushSubscriptionRepository()), - ReactionsController::class => new ReactionsController(static::reactionRepository()), + ReactionsController::class => new ReactionsController(static::reactionRepository(), static::channelRepository()), Auth::class => new Auth(static::userRepository(), static::auditLogger()), Profile::class => new Profile(static::userRepository()), CspReport::class => new CspReport(static::cspReportRepository()), @@ -90,6 +94,15 @@ public static function chatRepository(bool $getShared = true): ChatRepository return new ChatModel(); } + public static function channelRepository(bool $getShared = true): ChannelRepository + { + if ($getShared) { + return static::getSharedInstance('channelRepository'); + } + + return new ChannelModel(); + } + public static function userRepository(bool $getShared = true): UserRepository { if ($getShared) { diff --git a/app/Contracts/ChannelRepository.php b/app/Contracts/ChannelRepository.php new file mode 100644 index 0000000..262ad05 --- /dev/null +++ b/app/Contracts/ChannelRepository.php @@ -0,0 +1,38 @@ +> */ + public function listForUser(int $userId): array; + + /** @return array|null */ + public function findChannel(int $channelId): ?array; + + public function isMember(int $channelId, int $userId): bool; + + public function isCreator(int $channelId, int $userId): bool; + + public function createPublic(string $name, string $slug, ?string $topic, int $creatorId): int|false; + + public function join(int $channelId, int $userId): bool; + + public function leave(int $channelId, int $userId): bool; + + /** @param array{name?: string, slug?: string, topic?: ?string, archived_at?: ?string} $changes */ + public function updateChannel(int $channelId, array $changes): bool; + + public function findOrCreateDm(int $firstUserId, int $secondUserId): int|false; + + /** @return list */ + public function memberIds(int $channelId): array; + + public function markRead(int $channelId, int $userId): bool; + + public function messageChannelId(int $messageId): ?int; +} diff --git a/app/Contracts/ChatRepository.php b/app/Contracts/ChatRepository.php index de35277..038f525 100644 --- a/app/Contracts/ChatRepository.php +++ b/app/Contracts/ChatRepository.php @@ -10,7 +10,7 @@ interface ChatRepository * pagination: array * } */ - public function getMsgPaginated(int $page = 1, int $perPage = 10): array; + public function getMsgPaginated(int $page = 1, int $perPage = 10, ?int $channelId = null): array; /** * @return array{ @@ -25,7 +25,8 @@ public function searchMessages( ?int $to = null, int $page = 1, int $perPage = 10, + ?int $channelId = null, ): array; - public function insertMsg(string $name, string $message, int $current): int|bool; + public function insertMsg(string $name, string $message, int $current, ?int $channelId = null): int|bool; } diff --git a/app/Controllers/Api/V1/ChannelsController.php b/app/Controllers/Api/V1/ChannelsController.php new file mode 100644 index 0000000..8975c38 --- /dev/null +++ b/app/Controllers/Api/V1/ChannelsController.php @@ -0,0 +1,367 @@ +channelJson(['channels' => $this->channels->listForUser($this->userId())]); + } + + public function create(): ResponseInterface + { + $input = $this->payload(); + $error = $this->validateChannelFields($input, false); + if ($error !== null) { + return $this->error('validation', $error, 400); + } + + try { + $channelId = $this->channels->createPublic( + trim((string) $input['name']), + trim((string) $input['slug']), + $this->nullableTrimmedString($input['topic'] ?? null), + $this->userId(), + ); + } catch (Throwable) { + return $this->error('conflict', 'The channel slug is already in use.', 409); + } + + if ($channelId === false) { + return $this->error('database', 'The channel could not be created.', 500); + } + + return $this->channelJson(['channel' => $this->channels->findChannel($channelId)], 201); + } + + public function join(int $channelId): ResponseInterface + { + $channel = $this->channels->findChannel($channelId); + if ($channel === null) { + return $this->error('not_found', 'Channel not found.', 404); + } + if ($channel['channel_type'] !== 'public' || $channel['archived_at'] !== null) { + return $this->error('authorization', 'This channel cannot be joined.', 403); + } + + if (! $this->channels->join($channelId, $this->userId())) { + return $this->error('database', 'The channel could not be joined.', 500); + } + + return $this->channelJson(['channel' => $this->channels->findChannel($channelId)]); + } + + public function leave(int $channelId): ResponseInterface + { + $channel = $this->channels->findChannel($channelId); + if ($channel === null) { + return $this->error('not_found', 'Channel not found.', 404); + } + if ($channel['slug'] === 'general') { + return $this->error('validation', 'The #general channel cannot be left.', 400); + } + if ($channel['channel_type'] === 'dm') { + return $this->error('validation', 'Direct-message membership cannot be changed.', 400); + } + if (! $this->channels->isMember($channelId, $this->userId())) { + return $this->error('authorization', 'You are not a channel member.', 403); + } + + $this->channels->leave($channelId, $this->userId()); + + return $this->channelJson(['success' => true]); + } + + public function update(int $channelId): ResponseInterface + { + $channel = $this->channels->findChannel($channelId); + if ($channel === null) { + return $this->error('not_found', 'Channel not found.', 404); + } + if ($channel['channel_type'] !== 'public' || ! $this->channels->isCreator($channelId, $this->userId())) { + return $this->error('authorization', 'Only the channel creator can update this channel.', 403); + } + + $input = $this->payload(); + if ($input === []) { + return $this->error('validation', 'Provide at least one channel field to update.', 400); + } + $error = $this->validateChannelFields($input, true); + if ($error !== null) { + return $this->error('validation', $error, 400); + } + + $changes = []; + foreach (['name', 'slug'] as $field) { + if (array_key_exists($field, $input)) { + $changes[$field] = trim((string) $input[$field]); + } + } + if (array_key_exists('topic', $input)) { + $changes['topic'] = $this->nullableTrimmedString($input['topic']); + } + if (array_key_exists('archived', $input)) { + $archived = filter_var($input['archived'], FILTER_VALIDATE_BOOL, FILTER_NULL_ON_FAILURE); + if ($archived === null) { + return $this->error('validation', 'The archived field must be a boolean.', 400); + } + $changes['archived_at'] = $archived ? date('Y-m-d H:i:s') : null; + } + if ($changes === []) { + return $this->error('validation', 'No supported channel fields were provided.', 400); + } + + try { + $updated = $this->channels->updateChannel($channelId, $changes); + } catch (Throwable) { + return $this->error('conflict', 'The channel slug is already in use.', 409); + } + + if (! $updated) { + return $this->error('database', 'The channel could not be updated.', 500); + } + + return $this->channelJson(['channel' => $this->channels->findChannel($channelId)]); + } + + public function createDm(): ResponseInterface + { + $input = $this->payload(); + $targetUserId = $input['user_id'] ?? null; + if (is_int($targetUserId) || (is_string($targetUserId) && ctype_digit($targetUserId))) { + $user = $this->users->findUserById((int) $targetUserId); + } else { + $username = $input['username'] ?? null; + if (! is_string($username) || trim($username) === '') { + return $this->error('validation', 'A user_id or username is required.', 400); + } + $user = $this->users->findUserByUsername(trim($username)); + } + if ($user === null) { + return $this->error('not_found', 'User not found.', 404); + } + if ((int) $user['id'] === $this->userId()) { + return $this->error('validation', 'A direct message requires another user.', 400); + } + + $channelId = $this->channels->findOrCreateDm($this->userId(), (int) $user['id']); + if ($channelId === false) { + return $this->error('database', 'The direct message could not be created.', 500); + } + + return $this->channelJson(['channel' => $this->channels->findChannel($channelId)], 201); + } + + public function messages(int $channelId): ResponseInterface + { + $membershipError = $this->memberChannel($channelId); + if ($membershipError instanceof ResponseInterface) { + return $membershipError; + } + + $page = $this->positiveIntegerQuery('page', 1, PHP_INT_MAX); + $perPage = $this->positiveIntegerQuery('per_page', 10, 100); + if (is_string($page) || is_string($perPage)) { + return $this->error('validation', is_string($page) ? $page : $perPage, 400); + } + + $result = $this->messages->getMsgPaginated($page, $perPage, $channelId); + $this->channels->markRead($channelId, $this->userId()); + + return $this->channelJson($result); + } + + public function createMessage(int $channelId): ResponseInterface + { + $membershipError = $this->memberChannel($channelId); + if ($membershipError instanceof ResponseInterface) { + return $membershipError; + } + + $message = $this->payload()['message'] ?? null; + if (! is_string($message) || trim($message) === '' || mb_strlen($message) > 500) { + return $this->error('validation', 'Message must contain between 1 and 500 characters.', 400); + } + + $messageId = $this->messages->insertMsg((string) $this->getCurrentUsername(), esc(trim($message)), time(), $channelId); + if ($messageId === false) { + return $this->error('database', 'The message could not be created.', 500); + } + $this->channels->markRead($channelId, $this->userId()); + + return $this->channelJson([ + 'message' => [ + 'id' => (int) $messageId, + 'channel_id' => $channelId, + 'user' => $this->getCurrentUsername(), + 'msg' => esc(trim($message)), + ], + ], 201); + } + + public function searchMessages(int $channelId): ResponseInterface + { + $membershipError = $this->memberChannel($channelId); + if ($membershipError instanceof ResponseInterface) { + return $membershipError; + } + + $textInput = $this->optionalQueryString('text', 500); + $userInput = $this->optionalQueryString('user', 255); + $fromInput = $this->optionalTimestamp('from'); + $toInput = $this->optionalTimestamp('to'); + $page = $this->positiveIntegerQuery('page', 1, PHP_INT_MAX); + $perPage = $this->positiveIntegerQuery('per_page', 10, 100); + foreach ([$textInput, $userInput, $fromInput, $toInput] as $input) { + if ($input['error'] !== null) { + return $this->error('validation', $input['error'], 400); + } + } + if (is_string($page) || is_string($perPage)) { + return $this->error('validation', is_string($page) ? $page : $perPage, 400); + } + $text = $textInput['value']; + $user = $userInput['value']; + $from = $fromInput['value']; + $to = $toInput['value']; + if ($text === null && $user === null && $from === null && $to === null) { + return $this->error('validation', 'Provide at least one search filter.', 400); + } + if (is_int($from) && is_int($to) && $from > $to) { + return $this->error('validation', 'The from timestamp must be less than or equal to to.', 400); + } + + $result = $this->messages->searchMessages($text, $user, $from, $to, $page, $perPage, $channelId); + + return $this->channelJson([ + ...$result, + 'filters' => ['text' => $text, 'user' => $user, 'from' => $from, 'to' => $to], + ]); + } + + private function memberChannel(int $channelId): ?ResponseInterface + { + $channel = $this->channels->findChannel($channelId); + if ($channel === null) { + return $this->error('not_found', 'Channel not found.', 404); + } + if (! $this->channels->isMember($channelId, $this->userId())) { + return $this->error('authorization', 'You are not a channel member.', 403); + } + + return null; + } + + /** @return array */ + private function payload(): array + { + $json = $this->request->getJSON(true); + + return is_array($json) ? $json : $this->request->getPost(); + } + + /** @param array $input */ + private function validateChannelFields(array $input, bool $partial): ?string + { + if (! $partial || array_key_exists('name', $input)) { + if (! is_string($input['name'] ?? null) || trim((string) $input['name']) === '' || mb_strlen(trim((string) $input['name'])) > 80) { + return 'Channel name must contain between 1 and 80 characters.'; + } + } + if (! $partial || array_key_exists('slug', $input)) { + if (! is_string($input['slug'] ?? null) || preg_match('/^[a-z0-9-]{3,50}$/', trim((string) $input['slug'])) !== 1) { + return 'Channel slug must match ^[a-z0-9-]{3,50}$.'; + } + if (trim((string) $input['slug']) === 'general' && $partial) { + return 'The #general slug is reserved.'; + } + } + if (array_key_exists('topic', $input) && $input['topic'] !== null && (! is_string($input['topic']) || mb_strlen(trim($input['topic'])) > 200)) { + return 'Channel topic must not exceed 200 characters.'; + } + + return null; + } + + private function nullableTrimmedString(mixed $value): ?string + { + if (! is_string($value)) { + return null; + } + + $value = trim($value); + + return $value === '' ? null : $value; + } + + private function positiveIntegerQuery(string $name, int $default, int $maximum): int|string + { + $value = $this->request->getGet($name); + if ($value === null) { + return $default; + } + if (! is_string($value) || preg_match('/^[1-9]\d*$/', $value) !== 1 || (float) $value > $maximum) { + return "The {$name} parameter must be an integer between 1 and {$maximum}."; + } + + return (int) $value; + } + + /** @return array{value: ?string, error: ?string} */ + private function optionalQueryString(string $name, int $maximum): array + { + $value = $this->request->getGet($name); + if ($value === null) { + return ['value' => null, 'error' => null]; + } + if (! is_string($value) || trim($value) === '' || mb_strlen(trim($value)) > $maximum) { + return ['value' => null, 'error' => "The {$name} filter must be a non-empty string of at most {$maximum} characters."]; + } + + return ['value' => trim($value), 'error' => null]; + } + + /** @return array{value: ?int, error: ?string} */ + private function optionalTimestamp(string $name): array + { + $value = $this->request->getGet($name); + if ($value === null) { + return ['value' => null, 'error' => null]; + } + if (! is_string($value) || preg_match('/^(0|[1-9]\d*)$/', $value) !== 1 || (float) $value > PHP_INT_MAX) { + return ['value' => null, 'error' => "The {$name} filter must be a non-negative Unix timestamp."]; + } + + return ['value' => (int) $value, 'error' => null]; + } + + private function userId(): int + { + return (int) $this->getCurrentUserId(); + } + + private function error(string $type, string $message, int $status): ResponseInterface + { + return $this->channelJson(['error' => ['type' => $type, 'message' => $message]], $status); + } + + private function channelJson(mixed $data, int $status = 200): ResponseInterface + { + return $this->respondWithJson($data, $status)->setHeader('X-CSRF-TOKEN', csrf_hash()); + } +} diff --git a/app/Controllers/Api/V1/ReactionsController.php b/app/Controllers/Api/V1/ReactionsController.php index c24c57c..aebcabe 100644 --- a/app/Controllers/Api/V1/ReactionsController.php +++ b/app/Controllers/Api/V1/ReactionsController.php @@ -2,19 +2,22 @@ namespace App\Controllers\Api\V1; +use App\Contracts\ChannelRepository; use App\Contracts\ReactionRepository; use App\Controllers\BaseController; use CodeIgniter\HTTP\ResponseInterface; final class ReactionsController extends BaseController { - public function __construct(private readonly ReactionRepository $reactions) - { + public function __construct( + private readonly ReactionRepository $reactions, + private readonly ChannelRepository $channels, + ) { } public function list(int $messageId): ResponseInterface { - if (! $this->reactions->messageExists($messageId)) { + if (! $this->canAccessMessage($messageId)) { return $this->notFound(); } @@ -23,7 +26,7 @@ public function list(int $messageId): ResponseInterface public function create(int $messageId): ResponseInterface { - if (! $this->reactions->messageExists($messageId)) { + if (! $this->canAccessMessage($messageId)) { return $this->notFound(); } @@ -45,7 +48,7 @@ public function create(int $messageId): ResponseInterface public function delete(int $messageId, string $emoji): ResponseInterface { - if (! $this->reactions->messageExists($messageId)) { + if (! $this->canAccessMessage($messageId)) { return $this->notFound(); } @@ -102,6 +105,25 @@ private function currentUserId(): int return (int) $this->session->get('user_id'); } + private function canAccessMessage(int $messageId): bool + { + if (! $this->reactions->messageExists($messageId)) { + return false; + } + + $channelId = $this->channels->messageChannelId($messageId); + if ($channelId === null) { + return false; + } + + $userId = $this->currentUserId(); + if ($channelId === $this->channels->generalChannelId()) { + $this->channels->ensureGeneralMembership($userId); + } + + return $this->channels->isMember($channelId, $userId); + } + private function notFound(): ResponseInterface { return $this->reactionJson([ diff --git a/app/Database/Migrations/2026-09-16-000000_create_channels.php b/app/Database/Migrations/2026-09-16-000000_create_channels.php new file mode 100644 index 0000000..fd82760 --- /dev/null +++ b/app/Database/Migrations/2026-09-16-000000_create_channels.php @@ -0,0 +1,115 @@ +forge->addField([ + 'id' => ['type' => 'INT', 'constraint' => 11, 'unsigned' => true, 'auto_increment' => true], + 'name' => ['type' => 'VARCHAR', 'constraint' => 80], + 'slug' => ['type' => 'VARCHAR', 'constraint' => 50], + 'channel_type' => ['type' => 'VARCHAR', 'constraint' => 16, 'default' => 'public'], + 'topic' => ['type' => 'VARCHAR', 'constraint' => 200, 'null' => true], + 'created_by' => ['type' => 'INT', 'constraint' => 11, 'unsigned' => true, 'null' => true], + 'archived_at' => ['type' => 'DATETIME', 'null' => true], + 'created_at' => ['type' => 'DATETIME', 'null' => true], + 'updated_at' => ['type' => 'DATETIME', 'null' => true], + ]); + $this->forge->addKey('id', true); + $this->forge->addUniqueKey('slug'); + $this->forge->addKey(['channel_type', 'archived_at']); + $this->forge->addForeignKey('created_by', 'users', 'id', 'CASCADE', 'SET NULL'); + $this->forge->createTable('channels'); + + $now = date('Y-m-d H:i:s'); + $this->database()->table('channels')->insert([ + 'name' => 'General', + 'slug' => 'general', + 'channel_type' => 'public', + 'topic' => 'General conversation', + 'created_at' => $now, + 'updated_at' => $now, + ]); + $generalId = (int) $this->database()->insertID(); + + $this->forge->addField([ + 'channel_id' => ['type' => 'INT', 'constraint' => 11, 'unsigned' => true], + 'user_id' => ['type' => 'INT', 'constraint' => 11, 'unsigned' => true], + 'last_read_message_id' => ['type' => 'INT', 'constraint' => 7, 'null' => true], + 'joined_at' => ['type' => 'DATETIME', 'null' => true], + ]); + $this->forge->addKey(['channel_id', 'user_id'], true); + $this->forge->addKey('user_id'); + $this->forge->addForeignKey('channel_id', 'channels', 'id', 'CASCADE', 'CASCADE'); + $this->forge->addForeignKey('user_id', 'users', 'id', 'CASCADE', 'CASCADE'); + $this->forge->createTable('channel_members'); + + $users = $this->database()->prefixTable('users'); + $members = $this->database()->prefixTable('channel_members'); + $this->database()->query( + "INSERT INTO {$members} (channel_id, user_id, joined_at) SELECT ?, id, ? FROM {$users}", + [$generalId, $now], + ); + + $this->addMessageChannel($generalId); + + if ($this->database()->getPlatform() === 'MySQLi') { + $channels = $this->database()->prefixTable('channels'); + $this->database()->query("ALTER TABLE {$channels} CONVERT TO CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci"); + } + } + + public function down(): void + { + $messages = $this->database()->prefixTable('messages'); + + if ($this->database()->getPlatform() === 'MySQLi') { + $this->database()->query("ALTER TABLE {$messages} DROP FOREIGN KEY " . self::MESSAGE_CHANNEL_FK); + $this->database()->query("ALTER TABLE {$messages} DROP INDEX idx_messages_channel_time"); + $this->database()->query("ALTER TABLE {$messages} DROP COLUMN channel_id"); + } else { + $this->database()->query('DROP INDEX IF EXISTS idx_messages_channel_time'); + $this->database()->query("ALTER TABLE {$messages} DROP COLUMN channel_id"); + } + + $this->forge->dropTable('channel_members'); + $this->forge->dropTable('channels'); + } + + private function addMessageChannel(int $generalId): void + { + $messages = $this->database()->prefixTable('messages'); + + if ($this->database()->getPlatform() === 'MySQLi') { + $channels = $this->database()->prefixTable('channels'); + $this->database()->query("ALTER TABLE {$messages} ADD channel_id INT(11) UNSIGNED NULL AFTER id"); + $this->database()->query("UPDATE {$messages} SET channel_id = ? WHERE channel_id IS NULL", [$generalId]); + $this->database()->query("ALTER TABLE {$messages} MODIFY channel_id INT(11) UNSIGNED NOT NULL DEFAULT {$generalId}"); + $this->database()->query("ALTER TABLE {$messages} ADD INDEX idx_messages_channel_time (channel_id, time, id)"); + $this->database()->query( + "ALTER TABLE {$messages} ADD CONSTRAINT " . self::MESSAGE_CHANNEL_FK . " FOREIGN KEY (channel_id) REFERENCES {$channels}(id) ON DELETE RESTRICT ON UPDATE CASCADE", + ); + + return; + } + + $this->database()->query("ALTER TABLE {$messages} ADD COLUMN channel_id INTEGER NOT NULL DEFAULT {$generalId}"); + $this->database()->query("CREATE INDEX idx_messages_channel_time ON {$messages} (channel_id, time, id)"); + } + + private function database(): BaseConnection + { + if (! $this->db instanceof BaseConnection) { + throw new \LogicException('Channel migrations require a CodeIgniter base database connection.'); + } + + return $this->db; + } +} diff --git a/app/Libraries/ChatWebSocketServer.php b/app/Libraries/ChatWebSocketServer.php index 4476150..069b5b7 100644 --- a/app/Libraries/ChatWebSocketServer.php +++ b/app/Libraries/ChatWebSocketServer.php @@ -2,9 +2,11 @@ namespace App\Libraries; +use App\Contracts\ChannelRepository; use App\Contracts\ReactionRepository; use App\Contracts\UserRepository; use App\Helpers\WebSocketTokenHelper; +use App\Models\ChannelModel; use App\Models\ChatModel; use App\Models\MessageReactionModel; use App\Models\UserModel; @@ -78,6 +80,8 @@ class ChatWebSocketServer implements MessageComponentInterface private readonly ReactionRepository $reactions; + private readonly ChannelRepository $channels; + /** @var array> */ private array $reactionHistory = []; @@ -87,7 +91,7 @@ class ChatWebSocketServer implements MessageComponentInterface /** * Ephemeral typing state keyed by authenticated user ID. * - * @var array + * @var array */ private array $typingUsers = []; @@ -107,12 +111,14 @@ public function __construct( bool $requireAuth = true, ?UserRepository $users = null, ?ReactionRepository $reactions = null, + ?ChannelRepository $channels = null, ) { $this->clients = $clients ?? new SplObjectStorage(); $this->chatModel = $chatModel ?? new ChatModel(); $this->requireAuth = $requireAuth; $this->users = $users ?? new UserModel(); $this->reactions = $reactions ?? new MessageReactionModel(); + $this->channels = $channels ?? new ChannelModel(); $this->logServerStart(); } @@ -154,10 +160,12 @@ public function onOpen(ConnectionInterface $conn): void // Store the connection with metadata // We use SplObjectStorage->attach() to associate data with the connection + $channelId = $userId > 0 ? $this->channels->ensureGeneralMembership($userId) : $this->channels->generalChannelId(); $this->clients->attach($conn, [ 'user_id' => $userId, 'connected_at' => time(), 'authenticated' => $this->requireAuth ? true : ($userId > 0), + 'channel_id' => $channelId, ]); if ($this->requireAuth) { @@ -203,6 +211,10 @@ public function onMessage(ConnectionInterface $from, $msg) $messageType = $data['type'] ?? $data['action']; switch ($messageType) { + case 'channel_subscribe': + $this->handleChannelSubscribe($from, $data); + break; + case 'getMessages': $this->handleGetMessages($from, $data); break; @@ -247,8 +259,9 @@ public function onClose(ConnectionInterface $conn): void $this->clients->detach($conn); if (is_int($userId) && ! $this->isUserConnected($userId) && isset($this->typingUsers[$userId])) { + $channelId = $this->typingUsers[$userId]['channel_id']; unset($this->typingUsers[$userId]); - $this->broadcastTypingState(); + $this->broadcastTypingState(channelId: $channelId); } if (is_int($userId) && $userId > 0 && ! $this->isUserConnected($userId) && isset($this->presenceUsers[$userId])) { @@ -317,7 +330,7 @@ public function isUserConnected(int $userId): bool public function pruneInactiveTypers(?int $now = null): void { $now ??= time(); - $changed = false; + $changedChannels = []; foreach ($this->typingUsers as $userId => $typingUser) { if ($now - $typingUser['last_typed_at'] < self::TYPING_TTL_SECONDS) { @@ -325,11 +338,11 @@ public function pruneInactiveTypers(?int $now = null): void } unset($this->typingUsers[$userId]); - $changed = true; + $changedChannels[$typingUser['channel_id']] = true; } - if ($changed) { - $this->broadcastTypingState(); + foreach (array_keys($changedChannels) as $channelId) { + $this->broadcastTypingState(channelId: $channelId); } } @@ -447,6 +460,12 @@ private function handleGetMessages(ConnectionInterface $from, array $data): void $page = $data['page'] ?? 1; $perPage = $data['perPage'] ?? 10; $requestId = isset($data['requestId']) && is_int($data['requestId']) ? $data['requestId'] : null; + $channelId = $this->authorizedChannelId($from, $data['channel_id'] ?? null); + if ($channelId === null) { + $this->sendChannelError($from, 'CHANNEL_ACCESS_DENIED', 'Channel membership is required.'); + + return; + } if (isset($data['search'])) { if (! is_array($data['search'])) { @@ -469,6 +488,7 @@ private function handleGetMessages(ConnectionInterface $from, array $data): void $search['filters']['to'], is_int($page) ? max(1, $page) : 1, is_int($perPage) ? min(100, max(1, $perPage)) : 10, + $channelId, ); $from->send(json_encode([ @@ -478,6 +498,7 @@ private function handleGetMessages(ConnectionInterface $from, array $data): void 'messages' => $result['messages'], 'pagination' => $result['pagination'], 'filters' => $search['filters'], + 'channel_id' => $channelId, ], ])); @@ -485,7 +506,16 @@ private function handleGetMessages(ConnectionInterface $from, array $data): void } // Fetch messages from the database - $result = $this->chatModel->getMsgPaginated($page, $perPage); + $result = $this->chatModel->getMsgPaginated( + is_int($page) ? max(1, $page) : 1, + is_int($perPage) ? min(100, max(1, $perPage)) : 10, + $channelId, + ); + $connectionData = $this->clients[$from] ?? []; + $userId = $connectionData['user_id'] ?? 0; + if (is_int($userId) && $userId > 0) { + $this->channels->markRead($channelId, $userId); + } // Send messages back to the client $from->send(json_encode([ @@ -493,6 +523,7 @@ private function handleGetMessages(ConnectionInterface $from, array $data): void 'data' => [ 'messages' => $result['messages'], 'pagination' => $result['pagination'], + 'channel_id' => $channelId, ], ])); } @@ -563,6 +594,7 @@ private function handleTypingEvent(ConnectionInterface $from, array $data, strin $authenticatedUserId = $connectionData['user_id'] ?? 0; $claimedUserId = $data['user_id'] ?? null; $username = $data['username'] ?? null; + $channelId = $connectionData['channel_id'] ?? 0; if ( ($connectionData['authenticated'] ?? false) !== true @@ -573,6 +605,8 @@ private function handleTypingEvent(ConnectionInterface $from, array $data, strin || ! is_string($username) || trim($username) === '' || mb_strlen($username) > 255 + || ! is_int($channelId) + || ! $this->channels->isMember($channelId, $authenticatedUserId) ) { return; } @@ -583,7 +617,7 @@ private function handleTypingEvent(ConnectionInterface $from, array $data, strin } unset($this->typingUsers[$authenticatedUserId]); - $this->broadcastTypingState($from); + $this->broadcastTypingState($from, $channelId); return; } @@ -594,22 +628,29 @@ private function handleTypingEvent(ConnectionInterface $from, array $data, strin $this->typingUsers[$authenticatedUserId] = [ 'user_id' => $authenticatedUserId, 'username' => $username, + 'channel_id' => $channelId, 'last_typed_at' => time(), ]; if ($changed) { - $this->broadcastTypingState($from); + $this->broadcastTypingState($from, $channelId); } } - private function broadcastTypingState(?ConnectionInterface $except = null): void + private function broadcastTypingState(?ConnectionInterface $except = null, ?int $channelId = null): void { + if ($channelId === null && $except !== null && $this->clients->contains($except)) { + $channelId = $this->clients[$except]['channel_id'] ?? null; + } $users = array_values(array_map( static fn (array $typingUser): array => [ 'user_id' => $typingUser['user_id'], 'username' => $typingUser['username'], ], - $this->typingUsers, + array_filter( + $this->typingUsers, + static fn (array $typingUser): bool => $channelId === null || $typingUser['channel_id'] === $channelId, + ), )); usort($users, static fn (array $left, array $right): int => [$left['username'], $left['user_id']] <=> [$right['username'], $right['user_id']]); @@ -619,6 +660,9 @@ private function broadcastTypingState(?ConnectionInterface $except = null): void if ($except !== null && $client === $except) { continue; } + if ($channelId !== null && ($this->clients[$client]['channel_id'] ?? null) !== $channelId) { + continue; + } $client->send($payload); } @@ -703,17 +747,32 @@ private function presenceProfile(array $user, string $presence, ?string $lastSee */ private function handleSendMessage(ConnectionInterface $from, array $data, int $userId): void { - // Validate required fields - if (!isset($data['message']) || !isset($data['username'])) { + if (! $this->clients->contains($from)) { + return; + } + + $connectionData = $this->clients[$from]; + $channelId = $this->authorizedChannelId($from, $data['channel_id'] ?? null); + $message = $data['message'] ?? null; + $user = $this->users->findUserById($userId); + if ( + ($connectionData['authenticated'] ?? false) !== true + || $channelId === null + || $user === null + || ! is_string($message) + || trim($message) === '' + || mb_strlen($message) > 500 + ) { + $this->sendChannelError($from, 'INVALID_MESSAGE', 'An authenticated channel member and a valid message are required.'); + return; } - $username = $data['username']; - $message = $data['message']; + $username = (string) $user['username']; + $message = trim($message); $timestamp = Time::now()->getTimestamp(); - // Insert message into database - $messageId = $this->chatModel->insertMsg($username, $message, $timestamp); + $messageId = $this->chatModel->insertMsg($username, $message, $timestamp, $channelId); // Prepare the message data for broadcasting $messageData = [ @@ -723,13 +782,27 @@ private function handleSendMessage(ConnectionInterface $from, array $data, int $ 'user' => $username, 'msg' => $message, 'timestamp' => $timestamp, + 'channel_id' => $channelId, ], ]; - // Broadcast the message to ALL connected clients - // This is what makes the chat "real-time" - everyone sees new messages instantly + $memberIds = $this->channels->memberIds($channelId); + $readBy = []; foreach ($this->clients as $client) { - $client->send(json_encode($messageData)); + $clientData = $this->clients[$client]; + $clientUserId = $clientData['user_id'] ?? 0; + if (! is_int($clientUserId) || ! in_array($clientUserId, $memberIds, true)) { + continue; + } + if (($clientData['channel_id'] ?? null) === $channelId) { + $client->send(json_encode($messageData)); + $readBy[$clientUserId] = true; + } else { + $client->send(json_encode(['type' => 'channel_unread', 'channel_id' => $channelId])); + } + } + foreach (array_keys($readBy) as $readUserId) { + $this->channels->markRead($channelId, $readUserId); } } @@ -744,6 +817,7 @@ private function handleReaction(ConnectionInterface $from, array $data, string $ $userId = $connectionData['user_id'] ?? 0; $messageId = $data['message_id'] ?? 0; $emoji = $this->validReactionEmoji($data['emoji'] ?? null); + $channelId = is_int($messageId) ? $this->channels->messageChannelId($messageId) : null; if ( ($connectionData['authenticated'] ?? false) !== true || ! is_int($userId) @@ -752,6 +826,8 @@ private function handleReaction(ConnectionInterface $from, array $data, string $ || $messageId <= 0 || $emoji === null || ! $this->reactions->messageExists($messageId) + || $channelId === null + || ! $this->channels->isMember($channelId, $userId) ) { $this->sendReactionError($from, 'INVALID_REACTION', 'The reaction request is invalid.'); @@ -785,11 +861,71 @@ private function handleReaction(ConnectionInterface $from, array $data, string $ $payload = json_encode([ 'type' => 'reaction', 'message_id' => $messageId, + 'channel_id' => $channelId, ...$reaction, ], JSON_UNESCAPED_UNICODE); foreach ($this->clients as $client) { - $client->send($payload); + $clientData = $this->clients[$client]; + if (($clientData['channel_id'] ?? null) === $channelId) { + $client->send($payload); + } + } + } + + /** @param array $data */ + private function handleChannelSubscribe(ConnectionInterface $from, array $data): void + { + $channelId = $data['channel_id'] ?? null; + if (! is_int($channelId) || ! $this->clients->contains($from)) { + $this->sendChannelError($from, 'INVALID_CHANNEL', 'A numeric channel_id is required.'); + + return; + } + + $connectionData = $this->clients[$from]; + $userId = $connectionData['user_id'] ?? 0; + if ( + ($connectionData['authenticated'] ?? false) !== true + || ! is_int($userId) + || ! $this->channels->isMember($channelId, $userId) + ) { + $this->sendChannelError($from, 'CHANNEL_ACCESS_DENIED', 'Channel membership is required.'); + + return; + } + + $connectionData['channel_id'] = $channelId; + $this->clients[$from] = $connectionData; + $this->channels->markRead($channelId, $userId); + $from->send(json_encode(['type' => 'channel_subscribed', 'channel_id' => $channelId])); + } + + private function authorizedChannelId(ConnectionInterface $from, mixed $requestedChannelId): ?int + { + if (! $this->clients->contains($from)) { + return null; + } + + $connectionData = $this->clients[$from]; + $channelId = is_int($requestedChannelId) ? $requestedChannelId : ($connectionData['channel_id'] ?? null); + if (! is_int($channelId)) { + return null; + } + + $userId = $connectionData['user_id'] ?? 0; + if (($connectionData['authenticated'] ?? false) === true) { + return is_int($userId) && $this->channels->isMember($channelId, $userId) ? $channelId : null; } + + return $channelId === $this->channels->generalChannelId() ? $channelId : null; + } + + private function sendChannelError(ConnectionInterface $connection, string $code, string $message): void + { + $connection->send(json_encode([ + 'action' => 'error', + 'data' => ['code' => $code, 'message' => $message], + ])); } private function allowReaction(int $userId): bool diff --git a/app/Models/ChannelModel.php b/app/Models/ChannelModel.php new file mode 100644 index 0000000..38cf8f0 --- /dev/null +++ b/app/Models/ChannelModel.php @@ -0,0 +1,210 @@ +select('id')->where('slug', 'general')->first(); + if (! is_array($row)) { + throw new \LogicException('The #general channel is missing. Run database migrations.'); + } + + return (int) $row['id']; + } + + public function ensureGeneralMembership(int $userId): int + { + $channelId = $this->generalChannelId(); + $this->db->table('channel_members')->ignore(true)->insert([ + 'channel_id' => $channelId, + 'user_id' => $userId, + 'joined_at' => date('Y-m-d H:i:s'), + ]); + + return $channelId; + } + + public function listForUser(int $userId): array + { + $this->ensureGeneralMembership($userId); + $channels = $this->db->table('channels AS channels') + ->select('channels.*, membership.last_read_message_id, membership.user_id AS membership_user_id') + ->select('(CASE WHEN membership.user_id IS NULL THEN 0 ELSE (SELECT COUNT(*) FROM ' . $this->db->prefixTable('messages') . ' AS unread_messages WHERE unread_messages.channel_id = channels.id AND unread_messages.id > COALESCE(membership.last_read_message_id, 0)) END) AS unread_count', false) + ->join('channel_members AS membership', 'membership.channel_id = channels.id AND membership.user_id = ' . $userId, 'left') + ->groupStart() + ->where('channels.channel_type', 'public') + ->orWhere('membership.user_id', $userId) + ->groupEnd() + ->where('channels.archived_at', null) + ->orderBy("CASE WHEN channels.slug = 'general' THEN 0 ELSE 1 END", '', false) + ->orderBy('channels.channel_type', 'ASC') + ->orderBy('channels.name', 'ASC') + ->get() + ->getResultArray(); + + foreach ($channels as &$channel) { + $channel['id'] = (int) $channel['id']; + $channel['created_by'] = $channel['created_by'] === null ? null : (int) $channel['created_by']; + $channel['unread_count'] = (int) $channel['unread_count']; + $channel['is_member'] = $channel['membership_user_id'] !== null; + unset($channel['membership_user_id']); + $channel['members'] = $this->memberUsernames((int) $channel['id']); + } + unset($channel); + + return $channels; + } + + public function findChannel(int $channelId): ?array + { + $channel = $this->find($channelId); + + return is_array($channel) ? $channel : null; + } + + public function isMember(int $channelId, int $userId): bool + { + return $this->db->table('channel_members') + ->where('channel_id', $channelId) + ->where('user_id', $userId) + ->countAllResults() === 1; + } + + public function isCreator(int $channelId, int $userId): bool + { + return $this->where('id', $channelId)->where('created_by', $userId)->countAllResults() === 1; + } + + public function createPublic(string $name, string $slug, ?string $topic, int $creatorId): int|false + { + $this->db->transStart(); + $channelId = $this->insert([ + 'name' => $name, + 'slug' => $slug, + 'channel_type' => 'public', + 'topic' => $topic, + 'created_by' => $creatorId, + ]); + if ($channelId !== false) { + $this->join((int) $channelId, $creatorId); + } + $this->db->transComplete(); + + return $this->db->transStatus() && $channelId !== false ? (int) $channelId : false; + } + + public function join(int $channelId, int $userId): bool + { + $saved = $this->db->table('channel_members')->ignore(true)->insert([ + 'channel_id' => $channelId, + 'user_id' => $userId, + 'joined_at' => date('Y-m-d H:i:s'), + ]); + + return $saved && $this->isMember($channelId, $userId); + } + + public function leave(int $channelId, int $userId): bool + { + if ($channelId === $this->generalChannelId()) { + return false; + } + + return $this->db->table('channel_members') + ->where('channel_id', $channelId) + ->where('user_id', $userId) + ->delete(); + } + + public function updateChannel(int $channelId, array $changes): bool + { + return $this->update($channelId, $changes); + } + + public function findOrCreateDm(int $firstUserId, int $secondUserId): int|false + { + [$firstUserId, $secondUserId] = [min($firstUserId, $secondUserId), max($firstUserId, $secondUserId)]; + $slug = "dm-{$firstUserId}-{$secondUserId}"; + $existing = $this->select('id')->where('slug', $slug)->first(); + if (is_array($existing)) { + $channelId = (int) $existing['id']; + $this->join($channelId, $firstUserId); + $this->join($channelId, $secondUserId); + + return $channelId; + } + + $this->db->transStart(); + $this->db->table('channels')->ignore(true)->insert([ + 'name' => 'Direct message', + 'slug' => $slug, + 'channel_type' => 'dm', + 'created_by' => $firstUserId, + 'created_at' => date('Y-m-d H:i:s'), + 'updated_at' => date('Y-m-d H:i:s'), + ]); + $channel = $this->select('id')->where('slug', $slug)->first(); + if (! is_array($channel)) { + $this->db->transRollback(); + + return false; + } + $channelId = (int) $channel['id']; + $this->join($channelId, $firstUserId); + $this->join($channelId, $secondUserId); + $this->db->transComplete(); + + return $this->db->transStatus() ? $channelId : false; + } + + public function memberIds(int $channelId): array + { + return array_map( + static fn (array $row): int => (int) $row['user_id'], + $this->db->table('channel_members')->select('user_id')->where('channel_id', $channelId)->get()->getResultArray(), + ); + } + + public function markRead(int $channelId, int $userId): bool + { + $row = $this->db->table('messages')->selectMax('id', 'latest_id')->where('channel_id', $channelId)->get()->getRowArray(); + $latestId = isset($row['latest_id']) ? (int) $row['latest_id'] : null; + + return $this->db->table('channel_members') + ->where('channel_id', $channelId) + ->where('user_id', $userId) + ->update(['last_read_message_id' => $latestId]); + } + + public function messageChannelId(int $messageId): ?int + { + $row = $this->db->table('messages')->select('channel_id')->where('id', $messageId)->get()->getRowArray(); + + return is_array($row) ? (int) $row['channel_id'] : null; + } + + /** @return list */ + private function memberUsernames(int $channelId): array + { + return array_column( + $this->db->table('channel_members') + ->select('users.username') + ->join('users', 'users.id = channel_members.user_id') + ->where('channel_members.channel_id', $channelId) + ->orderBy('users.username', 'ASC') + ->get() + ->getResultArray(), + 'username', + ); + } +} diff --git a/app/Models/ChatModel.php b/app/Models/ChatModel.php index 74876ff..25f2272 100644 --- a/app/Models/ChatModel.php +++ b/app/Models/ChatModel.php @@ -22,7 +22,7 @@ class ChatModel extends Model implements ChatRepository { protected $table = 'messages'; protected $primaryKey = 'id'; - protected $allowedFields = ['user', 'msg', 'time']; + protected $allowedFields = ['user', 'msg', 'time', 'channel_id']; /** * Cache key for messages @@ -57,7 +57,7 @@ public function __construct( * * @return array */ - public function getMsgPaginated(int $page = 1, int $perPage = 10): array + public function getMsgPaginated(int $page = 1, int $perPage = 10, ?int $channelId = null): array { // Ensure page is at least 1 $page = max(1, (int)$page); @@ -66,7 +66,8 @@ public function getMsgPaginated(int $page = 1, int $perPage = 10): array $offset = ($page - 1) * $perPage; // Create a unique cache key based on the pagination parameters - $cacheKey = $this->cacheKey . '_page_' . $page . '_' . $perPage; + $cacheChannel = $channelId ?? 'general'; + $cacheKey = $this->cacheKey . '_channel_' . $cacheChannel . '_page_' . $page . '_' . $perPage; // Get the cache service $cache = $this->cache; @@ -76,12 +77,14 @@ public function getMsgPaginated(int $page = 1, int $perPage = 10): array // If not in the cache or cache expired, get from the database and store in the cache if ($result === null) { + $channelId ??= $this->generalChannelId(); // Get total count for pagination - $totalCount = $this->countAllResults(); + $totalCount = $this->where('channel_id', $channelId)->countAllResults(); // Use time index for ordering instead of id // This is more efficient for chat applications where time-based ordering is natural - $messages = $this->orderBy('time', 'DESC') + $messages = $this->where('channel_id', $channelId) + ->orderBy('time', 'DESC') ->limit($perPage, $offset) ->get() ->getResultArray(); @@ -149,6 +152,7 @@ public function searchMessages( ?int $to = null, int $page = 1, int $perPage = 10, + ?int $channelId = null, ): array { $page = max(1, $page); $perPage = max(1, $perPage); @@ -159,6 +163,9 @@ public function searchMessages( $messagesTable = $this->db->prefixTable($this->table); $bindings = []; $conditions = []; + $channelId ??= $this->generalChannelId(); + $conditions[] = 'messages.channel_id = ?'; + $bindings[] = $channelId; if ($this->db->getPlatform() === 'SQLite3' && $text !== null && $text !== '') { $searchTable = $this->db->prefixTable('messages_fts'); @@ -225,12 +232,13 @@ public function searchMessages( * * @return int|bool The insert ID if the insert was successful, or false on failure */ - public function insertMsg(string $name, string $message, int $current): int|bool + public function insertMsg(string $name, string $message, int $current, ?int $channelId = null): int|bool { $result = $this->insert([ 'user' => $name, 'msg' => $message, 'time' => $current, + 'channel_id' => $channelId ?? $this->generalChannelId(), ]); // If insert was successful, invalidate the cache @@ -259,8 +267,10 @@ public function getMsgByUserPaginated(string $username, int $page = 1, int $perP // Calculate offset $offset = ($page - 1) * $perPage; + $channelId = $this->generalChannelId(); + // Create a unique cache key based on the username and pagination parameters - $cacheKey = $this->cacheKey . '_user_' . md5($username) . '_page_' . $page . '_' . $perPage; + $cacheKey = $this->cacheKey . '_channel_' . $channelId . '_user_' . md5($username) . '_page_' . $page . '_' . $perPage; // Get the cache service $cache = $this->cache; @@ -271,10 +281,11 @@ public function getMsgByUserPaginated(string $username, int $page = 1, int $perP // If not in the cache or cache expired, get from the database and store in the cache if ($result === null) { // Get total count for pagination - $totalCount = $this->where('user', $username)->countAllResults(); + $totalCount = $this->where('channel_id', $channelId)->where('user', $username)->countAllResults(); // Use user index for filtering and time index for ordering - $messages = $this->where('user', $username) + $messages = $this->where('channel_id', $channelId) + ->where('user', $username) ->orderBy('time', 'DESC') ->limit($perPage, $offset) ->get() @@ -344,8 +355,10 @@ public function getMsgByTimeRangePaginated(int $startTime, int $endTime, int $pa // Calculate offset $offset = ($page - 1) * $perPage; + $channelId = $this->generalChannelId(); + // Create a unique cache key based on the time range and pagination parameters - $cacheKey = $this->cacheKey . '_time_' . $startTime . '_' . $endTime . '_page_' . $page . '_' . $perPage; + $cacheKey = $this->cacheKey . '_channel_' . $channelId . '_time_' . $startTime . '_' . $endTime . '_page_' . $page . '_' . $perPage; // Get the cache service $cache = $this->cache; @@ -356,12 +369,14 @@ public function getMsgByTimeRangePaginated(int $startTime, int $endTime, int $pa // If not in the cache or cache expired, get from the database and store in the cache if ($result === null) { // Get total count for pagination - $totalCount = $this->where('time >=', $startTime) + $totalCount = $this->where('channel_id', $channelId) + ->where('time >=', $startTime) ->where('time <=', $endTime) ->countAllResults(); // Use time index for filtering and ordering - $messages = $this->where('time >=', $startTime) + $messages = $this->where('channel_id', $channelId) + ->where('time >=', $startTime) ->where('time <=', $endTime) ->orderBy('time', 'DESC') ->limit($perPage, $offset) @@ -440,4 +455,14 @@ private function toSqliteFtsQuery(string $text): string // A quoted phrase treats user input as text rather than FTS5 syntax. return '"' . str_replace('"', '""', $text) . '"'; } + + private function generalChannelId(): int + { + $row = $this->db->table('channels')->select('id')->where('slug', 'general')->get()->getRowArray(); + if (! is_array($row)) { + throw new \LogicException('The #general channel is missing. Run database migrations.'); + } + + return (int) $row['id']; + } } diff --git a/app/Views/chat/svelteView.php b/app/Views/chat/svelteView.php index 8582a5c..855ec30 100644 --- a/app/Views/chat/svelteView.php +++ b/app/Views/chat/svelteView.php @@ -76,6 +76,8 @@ update: "", // GET endpoint for fetching messages (JSON format) messagesApi: "", + channels: "", + dms: "", pushSubscriptions: "", profile: "", profileAvatar: "", diff --git a/app/Views/chat/vueView.php b/app/Views/chat/vueView.php index fde390b..60bb471 100644 --- a/app/Views/chat/vueView.php +++ b/app/Views/chat/vueView.php @@ -33,6 +33,8 @@ window.CHAT_ROUTES = { update: "", messagesApi: "", + channels: "", + dms: "", pushSubscriptions: "", profile: "", profileAvatar: "", diff --git a/docs/api-versioning.md b/docs/api-versioning.md index 8158cd4..aefdd58 100644 --- a/docs/api-versioning.md +++ b/docs/api-versioning.md @@ -31,6 +31,10 @@ All endpoints use the existing session authentication and CSRF rules. The canonical payload and error envelopes are documented in [`public-api.md`](public-api.md). +Version 1 also includes authenticated channel and direct-message endpoints. +Their validation, membership rules, and WebSocket events are documented in +[`channels.md`](channels.md). + ## Current deprecations `/chat/jsonBackend`, `/chat/vueApi`, and `/chat/svelteApi` are compatibility diff --git a/docs/channels.md b/docs/channels.md new file mode 100644 index 0000000..2a393f4 --- /dev/null +++ b/docs/channels.md @@ -0,0 +1,109 @@ +# Channels and direct messages + +Authenticated users can organize conversations into public channels and private +two-person direct messages. Existing HTML, XML, and JSON clients remain pinned +to `#general`; they do not need to send a channel identifier. + +## Data model + +- `channels` stores public channels and direct-message conversations. Public + slugs match `^[a-z0-9-]{3,50}$`; topics are optional and limited to 200 + characters. +- `channel_members` stores membership and the last-read message identifier used + for unread counts. +- `messages.channel_id` is required and indexed with message time and ID. +- The migration creates `#general`, enrolls existing users, and assigns every + existing message to it. New users are enrolled when they first use a channel + endpoint or open an authenticated WebSocket connection. + +`#general` cannot be left. A public channel's creator may rename it, change its +topic or slug, archive it, and restore it. This first version intentionally has +no granular roles or moderation permissions. + +## HTTP API + +All routes require session authentication. Mutating requests also require the +normal CSRF token and write-rate limit. + +| Method | Path | Purpose | +| --- | --- | --- | +| `GET` | `/api/v1/channels` | List active public channels plus the current user's DMs, memberships, and unread counts | +| `POST` | `/api/v1/channels` | Create a public channel and join its creator | +| `PATCH` | `/api/v1/channels/{id}` | Creator-only name, slug, topic, and archive changes | +| `POST` | `/api/v1/channels/{id}/join` | Join an active public channel | +| `POST` | `/api/v1/channels/{id}/leave` | Leave a public channel other than `#general` | +| `POST` | `/api/v1/dms` | Find or create an idempotent two-person DM by user ID | +| `GET` | `/api/v1/channels/{id}/messages?page=1&per_page=10` | Read member-scoped messages and mark them read | +| `POST` | `/api/v1/channels/{id}/messages` | Post a member-scoped message | +| `GET` | `/api/v1/channels/{id}/messages/search` | Search messages within one member-scoped channel | + +Create a public channel: + +```json +{ + "name": "Project Room", + "slug": "project-room", + "topic": "Planning and delivery" +} +``` + +Start or reopen a direct message (the username fallback is also accepted by the +built-in clients): + +```json +{ + "user_id": 42 +} +``` + +DM slugs are canonicalized from the two user IDs, so reversing the participants +returns the same channel. Non-members receive `403` for channel messages, +search, and direct-message content. + +## WebSocket protocol + +Authenticated connections begin subscribed to `#general`. Switch conversations +before reading or sending: + +```json +{ "type": "channel_subscribe", "channel_id": 12 } +``` + +The server confirms with `channel_subscribed`. `getMessages`, `sendMessage`, +typing, and reaction events are restricted to channel members. A new message is +delivered in full only to member connections currently subscribed to that +channel. Other connected members receive: + +```json +{ "type": "channel_unread", "channel_id": 12 } +``` + +Clients should include `channel_id` in `getMessages` and `sendMessage`. The +server derives the sender's username from the authenticated user and ignores a +client-supplied identity. + +## Deploying the migration + +The SQLite migration adds the required column with the new `#general` ID as its +default. MySQL uses an in-place sequence: add the column as nullable, backfill +existing rows, make it required with the `#general` default, then add the index +and foreign key. + +For a MySQL `messages` table with roughly 100,000 rows or more, schedule a +maintenance window or test an online-schema-change process on a production-sized +copy first. `ALTER TABLE` may rebuild the table or hold a metadata lock, while +the backfill creates write load and transaction-log growth. Before deployment: + +1. Verify a current backup and available disk space. +2. Measure the backfill and both `ALTER TABLE` operations on representative + hardware. +3. Stop or drain message writers if the selected MySQL version cannot perform + the changes online. +4. Monitor metadata-lock waits, replication lag, transaction logs, and free + space throughout the migration. +5. Confirm that all old messages point to `#general` and that + `idx_messages_channel_time` exists before restoring normal traffic. + +Rollback removes the message foreign key and index before dropping +`messages.channel_id`, `channel_members`, and `channels`. Rolling back discards +all channel assignments and should only be used with an explicit recovery plan. diff --git a/e2e/support/chat-scenarios.ts b/e2e/support/chat-scenarios.ts index f1d8762..eb08160 100644 --- a/e2e/support/chat-scenarios.ts +++ b/e2e/support/chat-scenarios.ts @@ -47,6 +47,10 @@ export function chatScenario(scenario: ChatScenario): void { await assertModernChatTyping(page, scenario.messageInput); await assertModernChatSearch(page); await assertModernChatReactions(page); + await expect(page.getByRole('navigation', { name: 'Channels and direct messages' })).toBeVisible(); + await expect(page.getByRole('button', { name: '# General' })).toHaveAttribute('aria-current', 'page'); + await expect(page.getByLabel('Start a direct message')).toBeVisible(); + await expect(page).toHaveURL(/channel=general/); await expect(page.locator('link[rel="manifest"]')).toHaveAttribute('href', '/manifest.webmanifest'); await page.context().setOffline(true); await expect(page.getByText('You’re offline.')).toBeVisible(); @@ -394,6 +398,7 @@ async function installModernChatWebSocket(page: Page): Promise { }, ], pagination: { hasNext: false }, + channel_id: request.channel_id, }, }; } else if (request.action === 'sendMessage') { @@ -404,6 +409,7 @@ async function installModernChatWebSocket(page: Page): Promise { user: request.username, msg: request.message, timestamp: Math.floor(Date.now() / 1000), + channel_id: request.channel_id, }, }; } diff --git a/src/svelte/App.svelte b/src/svelte/App.svelte index d216a3f..5b31fd0 100644 --- a/src/svelte/App.svelte +++ b/src/svelte/App.svelte @@ -132,6 +132,107 @@ const requestedProfiles = new Set(); let reactions = $state({}); let reactionPickerMessageId = $state(null); + let channels = $state([]); + let activeChannelId = $state(null); + let dmUsername = $state(''); + + async function initializeChat() { + await loadChannels(); + if (activeChannelId === null) return; + connectWebSocket(); + } + + async function loadChannels() { + try { + const response = await fetch(config.chatRoutes.channels); + if (!response.ok) throw new Error('Channel request failed'); + channels = (await response.json()).channels || []; + const requestedSlug = new URL(window.location.href).searchParams.get('channel'); + const requested = channels.find((channel) => channel.slug === requestedSlug); + const selected = + (requested?.is_member ? requested : null) || + channels.find((channel) => channel.slug === 'general') || + channels[0]; + if (selected && activeChannelId === null) { + activeChannelId = Number(selected.id); + const url = new URL(window.location.href); + url.searchParams.set('channel', selected.slug); + window.history.replaceState({}, '', url); + } + } catch { + error = 'Channels could not be loaded.'; + } + } + + function dmLabel(channel) { + return (channel.members || []).find((member) => member !== config.username) || 'Direct message'; + } + + async function selectChannel(channel) { + if (Number(channel.id) === Number(activeChannelId)) return; + if (!channel.is_member) { + try { + const response = await fetch(`${config.chatRoutes.channels}/${channel.id}/join`, { + method: 'POST', + headers: { 'X-CSRF-TOKEN': document.querySelector('meta[name="csrf-token"]')?.getAttribute('content') || '' }, + }); + refreshCsrfToken(response); + if (!response.ok) throw new Error('Channel join failed'); + channel.is_member = true; + } catch { + error = 'The channel could not be joined.'; + return; + } + } + stopTyping(); + activeChannelId = Number(channel.id); + channel.unread_count = 0; + messages = []; + currentPage = 1; + loading = true; + clearSearchState(); + const url = new URL(window.location.href); + url.searchParams.set('channel', channel.slug); + window.history.replaceState({}, '', url); + if (webSocketConnected) { + webSocket.send(JSON.stringify({ type: 'channel_subscribe', channel_id: activeChannelId })); + } + loadMessages(); + } + + function clearSearchState() { + cancelSearchRequest(); + search = { text: '', user: '', from: '', to: '' }; + searchActive = false; + searchLoading = false; + searchHasLiveUpdates = false; + searchError = ''; + activeSearchKey = ''; + } + + async function createDm() { + const username = dmUsername.trim(); + if (!username) return; + try { + const response = await fetch(config.chatRoutes.dms, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'X-CSRF-TOKEN': document.querySelector('meta[name="csrf-token"]')?.getAttribute('content') || '', + }, + body: JSON.stringify({ username }), + }); + refreshCsrfToken(response); + const payload = await response.json(); + if (!response.ok) throw new Error(payload.error?.message || 'Direct message failed'); + dmUsername = ''; + await loadChannels(); + const channel = channels.find((item) => Number(item.id) === Number(payload.channel.id)); + if (channel) selectChannel(channel); + } catch (err) { + error = err.message || 'The direct message could not be opened.'; + } + } function handleOwnProfile(profile) { displayName = profile.display_name || config.username; @@ -299,7 +400,7 @@ unsubscribePwa = subscribePwa((state) => { pwa = state; }); - connectWebSocket(); + void initializeChat(); // Add cleanup listener for page unload window.addEventListener('beforeunload', cleanUp); @@ -389,6 +490,10 @@ reconnectInterval = null; } + if (activeChannelId !== null) { + webSocket.send(JSON.stringify({ type: 'channel_subscribe', channel_id: activeChannelId })); + } + // Load initial messages once connected loadMessages(); warmMessagesCache(); @@ -410,10 +515,16 @@ applyReactionUpdate(data); return; } + if (data.type === 'channel_unread') { + const channel = channels.find((item) => Number(item.id) === Number(data.channel_id)); + if (channel) channel.unread_count = Number(channel.unread_count || 0) + 1; + return; + } // Handle different message types (actions) from the server switch (data.action) { case 'messages': + if (Number(data.data.channel_id) !== Number(activeChannelId)) break; if (searchActive) { break; } @@ -432,6 +543,7 @@ break; case 'newMessage': + if (Number(data.data.channel_id) !== Number(activeChannelId)) break; // Handle new message broadcast from another user if (searchActive) { searchHasLiveUpdates = true; @@ -559,6 +671,11 @@ * Load messages - prefers WebSocket, falls back to HTTP. */ function loadMessages() { + if (activeChannelId === null) { + loading = false; + return; + } + searchActive = false; searchHasLiveUpdates = false; searchLoading = false; @@ -580,6 +697,7 @@ action: 'getMessages', page: currentPage, perPage: 10, + channel_id: activeChannelId, }), ); } @@ -589,9 +707,17 @@ * Used when WebSocket is not available. */ async function loadMessagesHttp() { + if (activeChannelId === null) { + loading = false; + return false; + } + try { const response = await fetch( - userScopedMessagesUrl(`${config.chatRoutes.api}?page=${currentPage}&per_page=10`, config.userId), + userScopedMessagesUrl( + `${config.chatRoutes.channels}/${activeChannelId}/messages?page=${currentPage}&per_page=10`, + config.userId, + ), ); const data = await response.json(); if (!response.ok) throw new Error('Message request failed'); @@ -613,8 +739,15 @@ } async function warmMessagesCache() { + if (activeChannelId === null) return; + try { - await fetch(userScopedMessagesUrl(`${config.chatRoutes.api}?page=1&per_page=10`, config.userId)); + await fetch( + userScopedMessagesUrl( + `${config.chatRoutes.channels}/${activeChannelId}/messages?page=1&per_page=10`, + config.userId, + ), + ); } catch { // The live WebSocket remains authoritative; cache warming is best effort. } @@ -624,6 +757,7 @@ * Load more messages (pagination). */ function loadMoreMessages() { + if (activeChannelId === null) return; if (loadingMore) return; loadingMore = true; @@ -646,6 +780,7 @@ action: 'getMessages', page: currentPage, perPage: 10, + channel_id: activeChannelId, }), ); @@ -661,9 +796,17 @@ * HTTP fallback for loading more messages. */ async function loadMoreMessagesHttp() { + if (activeChannelId === null) { + loadingMore = false; + return; + } + try { const response = await fetch( - userScopedMessagesUrl(`${config.chatRoutes.api}?page=${currentPage}&per_page=10`, config.userId), + userScopedMessagesUrl( + `${config.chatRoutes.channels}/${activeChannelId}/messages?page=${currentPage}&per_page=10`, + config.userId, + ), ); const data = await response.json(); @@ -733,6 +876,8 @@ } function executeSearch(page = 1, append = false) { + if (activeChannelId === null) return; + const filters = normalizedSearch(); if (!Object.values(filters).some((value) => value !== '')) { clearSearch(); @@ -775,6 +920,7 @@ page, perPage: 10, search: compactSearch(filters), + channel_id: activeChannelId, }), ); @@ -899,7 +1045,7 @@ } function searchEndpoint() { - return `${config.chatRoutes.api.replace(/\/$/, '')}/search`; + return `${config.chatRoutes.channels}/${activeChannelId}/messages/search`; } function handleTypingInput(event) { @@ -966,6 +1112,11 @@ // Clear previous errors error = ''; + if (activeChannelId === null) { + error = 'Choose a conversation before sending a message.'; + return; + } + // Validate message if (!message.trim()) { error = 'Message is required'; @@ -988,6 +1139,7 @@ action: 'sendMessage', username: config.username, message: message, + channel_id: activeChannelId, }), ); @@ -1004,25 +1156,22 @@ * HTTP fallback for sending messages. */ async function sendMessageHttp() { + if (activeChannelId === null) { + sending = false; + return; + } + let fetchCompleted = false; try { - const formData = new FormData(); - formData.append('message', message); - formData.append('action', 'postmsg'); - - // Get CSRF token from meta tag - const csrfMeta = document.querySelector('meta[name="csrf-token"]'); - if (csrfMeta) { - formData.append(config.csrfTokenName, csrfMeta.getAttribute('content')); - } - - const response = await fetch(config.chatRoutes.update, { + const response = await fetch(`${config.chatRoutes.channels}/${activeChannelId}/messages`, { method: 'POST', - body: formData, headers: { - 'X-Requested-With': 'XMLHttpRequest', + 'Content-Type': 'application/json', + 'X-CSRF-TOKEN': document.querySelector('meta[name="csrf-token"]')?.getAttribute('content') || '', }, + body: JSON.stringify({ message }), }); + refreshCsrfToken(response); fetchCompleted = true; const data = await response.json(); @@ -1285,6 +1434,44 @@ {/if} + +

Search messages

@@ -1544,6 +1731,65 @@