feat: add Reverb realtime messaging

2026-03-21 12:51:59 +01:00
parent 60f78e8235
commit e8b5edf5d2
45 changed files with 3609 additions and 339 deletions
--- a/app/Services/Messaging/ConversationStateService.php
+++ b/app/Services/Messaging/ConversationStateService.php
@@ -0,0 +1,98 @@
+<?php
+
+namespace App\Services\Messaging;
+
+use App\Events\ConversationUpdated;
+use App\Events\MessageRead;
+use App\Models\Conversation;
+use App\Models\ConversationParticipant;
+use App\Models\Message;
+use App\Models\User;
+use Illuminate\Support\Facades\Cache;
+use Illuminate\Support\Facades\DB;
+
+class ConversationStateService
+{
+    public function activeParticipantIds(Conversation|int $conversation): array
+    {
+        $conversationId = $conversation instanceof Conversation ? $conversation->id : $conversation;
+
+        return ConversationParticipant::query()
+            ->where('conversation_id', $conversationId)
+            ->whereNull('left_at')
+            ->pluck('user_id')
+            ->map(fn ($id) => (int) $id)
+            ->all();
+    }
+
+    public function touchConversationCachesForUsers(array $userIds): void
+    {
+        foreach (array_unique($userIds) as $userId) {
+            if (! $userId) {
+                continue;
+            }
+
+            $versionKey = "messages:conversations:version:{$userId}";
+            Cache::add($versionKey, 1, now()->addDay());
+            Cache::increment($versionKey);
+        }
+    }
+
+    public function markConversationRead(Conversation $conversation, User $user, ?int $messageId = null): ConversationParticipant
+    {
+        /** @var ConversationParticipant $participant */
+        $participant = ConversationParticipant::query()
+            ->where('conversation_id', $conversation->id)
+            ->where('user_id', $user->id)
+            ->whereNull('left_at')
+            ->firstOrFail();
+
+        $lastReadableMessage = Message::query()
+            ->where('conversation_id', $conversation->id)
+            ->whereNull('deleted_at')
+            ->where('sender_id', '!=', $user->id)
+            ->when($messageId, fn ($query) => $query->where('id', '<=', $messageId))
+            ->orderByDesc('id')
+            ->first();
+
+        $readAt = now();
+
+        $participant->update([
+            'last_read_at' => $readAt,
+            'last_read_message_id' => $lastReadableMessage?->id,
+        ]);
+
+        if ($lastReadableMessage) {
+            $messageReads = Message::query()
+                ->select(['id'])
+                ->where('conversation_id', $conversation->id)
+                ->whereNull('deleted_at')
+                ->where('sender_id', '!=', $user->id)
+                ->where('id', '<=', $lastReadableMessage->id)
+                ->get()
+                ->map(fn (Message $message) => [
+                    'message_id' => $message->id,
+                    'user_id' => $user->id,
+                    'read_at' => $readAt,
+                ])
+                ->all();
+
+            if (! empty($messageReads)) {
+                DB::table('message_reads')->upsert($messageReads, ['message_id', 'user_id'], ['read_at']);
+            }
+        }
+
+        $participantIds = $this->activeParticipantIds($conversation);
+        $this->touchConversationCachesForUsers($participantIds);
+
+        DB::afterCommit(function () use ($conversation, $participant, $user): void {
+            event(new MessageRead($conversation, $participant, $user));
+
+            foreach ($this->activeParticipantIds($conversation) as $participantId) {
+                event(new ConversationUpdated($participantId, $conversation, 'message.read'));
+            }
+        });
+
+        return $participant->fresh(['user']);
+    }
+}