Laravel Reverb

<- pusher:connection_established  {"socket_id":"834043982.307151597","activity_timeout":30}
-> POST /api/broadcasting/auth {socket_id, channel_name: "presence-chat.general"}  200 {auth, channel_data}
-> pusher:subscribe {"channel":"presence-chat.general","auth":"<app key>:<hmac>","channel_data":"{\"user_id\":\"1\",...}"}
<- pusher_internal:subscription_succeeded presence-chat.general
   {"presence":{"count":1,"ids":["1"],"hash":{"1":{"userId":"1","name":"Alice"}}}}
<- pusher_internal:member_added presence-chat.general {"user_id":"2","user_info":{"userId":"2","name":"Bob"}}
<- message presence-chat.general
   {"v":1,"type":"message","id":"01JAB3W7Q8Z4N5R66FFL8X1000","room":"general","ts":1792871801210,
    "payload":{"from":{"userId":"1","name":"Alice"},"text":"Hello, Chatterbox","replyTo":null,"attachments":[],"seq":1,"edited":false}}
<- client-typing presence-chat.general {"state":"start"}
<- pusher_internal:member_removed presence-chat.general {"user_id":"1"}

FROM php:8.4-cli
RUN apt-get update && apt-get install -y --no-install-recommends git unzip libzip-dev && rm -rf /var/lib/apt/lists/* \
 && docker-php-ext-install pcntl zip pdo_mysql
COPY --from=composer:2.10.3 /usr/bin/composer /usr/local/bin/composer
WORKDIR /app

composer create-project laravel/laravel . --no-interaction
composer require laravel/reverb predis/predis -W      # reverb 1.12.0, predis 3.7.0
php artisan install:api                               # Sanctum 4 and the personal_access_tokens table
php artisan install:broadcasting --reverb --without-node --no-interaction

->withBroadcasting(__DIR__.'/../routes/channels.php', ['prefix' => 'api', 'middleware' => ['api', 'auth:sanctum']])

name: g3-laravel-reverb
x-laravel: &laravel
  build: .
  working_dir: /app
  volumes: [".:/app"]
  environment: &laravel-env
    REDIS_HOST: redis
    REDIS_CLIENT: predis
    CACHE_STORE: redis
    DB_CONNECTION: mysql
    DB_HOST: db
    DB_DATABASE: chatterbox
    DB_USERNAME: chatterbox
    DB_PASSWORD: change-me
    REVERB_HOST: reverb1                  # where Laravel POSTs events (server side)
    REVERB_PORT: "8080"
    REVERB_SCHEME: http
    REVERB_SCALING_ENABLED: ${SCALING:-true}
    REVERB_ALLOWED_ORIGINS: chat.example.test
    PHP_CLI_SERVER_WORKERS: "4"

services:
  db:
    image: mysql:8.4
    environment:
      MYSQL_DATABASE: chatterbox
      MYSQL_USER: chatterbox
      MYSQL_PASSWORD: change-me
      MYSQL_ROOT_PASSWORD: change-me-root
    healthcheck:
      test: ["CMD", "mysqladmin", "ping", "-h", "127.0.0.1", "-pchange-me-root"]
      interval: 3s
      retries: 30
  redis:
    image: redis:8.8.3-alpine
    ports: ["6430:6379"]
  app:
    <<: *laravel
    command: php -S 0.0.0.0:8000 -t public
    ports: ["8230:8000"]
    depends_on:
      db: { condition: service_healthy }
      reverb1: { condition: service_started }
  reverb1:
    <<: *laravel
    command: php artisan reverb:start --host=0.0.0.0 --port=8080 --debug
    ports: ["8231:8080"]
    depends_on: [redis]
  reverb2:
    <<: *laravel
    command: php artisan reverb:start --host=0.0.0.0 --port=8080
    ports: ["8232:8080"]
    depends_on: [redis]

<?php

use Illuminate\Database\Migrations\Migration;
use Illuminate\Database\Schema\Blueprint;
use Illuminate\Support\Facades\Schema;

return new class extends Migration
{
    public function up(): void
    {
        Schema::create('chat_rooms', function (Blueprint $t) {
            $t->string('name', 64)->primary();
            $t->unsignedBigInteger('last_seq')->default(0);      // one counter per room
        });

        Schema::create('chat_messages', function (Blueprint $t) {
            $t->string('id', 36)->primary();                     // the client id is the message id forever
            $t->string('room', 64);
            $t->foreignId('user_id');
            $t->unsignedBigInteger('seq');
            $t->text('text');
            $t->string('reply_to', 36)->nullable();
            $t->json('attachments');
            $t->unsignedBigInteger('created_ms');
            $t->unsignedBigInteger('edited_ms')->nullable();
            $t->unique(['room', 'seq']);
        });

        Schema::create('chat_reactions', function (Blueprint $t) {
            $t->string('target', 36);
            $t->foreignId('user_id');
            $t->string('emoji', 32);
            $t->primary(['target', 'user_id', 'emoji']);
        });

        Schema::create('chat_acks', function (Blueprint $t) {    // idempotency: (user, id) -> original ack
            $t->foreignId('user_id');
            $t->string('id', 36);
            $t->text('ack');
            $t->unsignedBigInteger('created_ms');
            $t->primary(['user_id', 'id']);
        });
    }

    public function down(): void
    {
        foreach (['chat_acks', 'chat_reactions', 'chat_messages', 'chat_rooms'] as $table) {
            Schema::dropIfExists($table);
        }
    }
};

<?php

namespace App\Chat;

use Illuminate\Support\Str;

final class Envelope
{
    public static function nowMs(): int
    {
        return (int) floor(microtime(true) * 1000);
    }

    public static function make(string $type, ?string $id, ?string $room, array $payload, ?int $ts = null): array
    {
        return [
            'v' => 1, 'type' => $type, 'id' => $id ?? 's_'.Str::upper((string) Str::ulid()),
            'room' => $room, 'ts' => $ts ?? self::nowMs(), 'payload' => (object) $payload,
        ];
    }

    public static function ack(string $ref, ?string $room, ?int $seq, array $result = [], bool $dup = false): array
    {
        return self::make('ack', null, $room, ['ref' => $ref, 'ok' => true, 'seq' => $seq, 'dup' => $dup, 'result' => (object) $result]);
    }
}

<?php

namespace App\Chat;

use RuntimeException;

final class ChatError extends RuntimeException
{
    private const TABLE = [
        'bad_request' => [400, false], 'unauthorized' => [401, false], 'forbidden' => [403, false],
        'not_found' => [404, false], 'conflict' => [409, false], 'too_large' => [413, false],
        'unsupported_version' => [426, false], 'rate_limited' => [429, true], 'internal' => [500, true],
        'unavailable' => [503, true],
    ];

    public function __construct(public readonly string $codeName, string $message)
    {
        parent::__construct($message);
    }

    public function status(): int { return self::TABLE[$this->codeName][0]; }

    public function retryable(): bool { return self::TABLE[$this->codeName][1]; }
}

<?php

namespace App\Chat;

use App\Models\User;
use Illuminate\Support\Str;

final class Rooms
{
    public static function valid(string $room): bool
    {
        return preg_match('/^[a-z0-9-]{1,64}$/', $room) === 1;
    }

    /** Public rooms are open to every signed-in user; dm-<a>-<b> only to the two users it names. */
    public static function canAccess(User $user, string $room): bool
    {
        if (! self::valid($room)) {
            return false;
        }
        if (str_starts_with($room, 'dm-')) {
            $ids = explode('-', Str::after($room, 'dm-'));

            return count($ids) === 2 && in_array((string) $user->id, $ids, true);
        }

        return true;
    }
}

<?php

namespace App\Events;

use Illuminate\Broadcasting\PresenceChannel;
use Illuminate\Contracts\Broadcasting\ShouldBroadcastNow;
use Illuminate\Foundation\Events\Dispatchable;

/** One Chatterbox envelope (message, edit or react) pushed to every member of a room. */
class ChatEnvelope implements ShouldBroadcastNow   // sent during the request: seq order is send order
{
    use Dispatchable;

    public function __construct(public string $room, public array $envelope) {}

    public function broadcastOn(): PresenceChannel
    {
        return new PresenceChannel('chat.'.$this->room);
    }

    public function broadcastAs(): string
    {
        return $this->envelope['type'];     // Echo listens for ".message", ".edit", ".react"
    }

    public function broadcastWith(): array
    {
        return $this->envelope;             // the whole envelope is the event data
    }
}

<?php

use App\Chat\Rooms;
use App\Models\User;
use Illuminate\Support\Facades\Broadcast;

// presence-chat.{room}: the member list is the Chatterbox member list
Broadcast::channel('chat.{room}', function (User $user, string $room) {
    return Rooms::canAccess($user, $room)
        ? ['userId' => (string) $user->id, 'name' => $user->name]
        : false;
});

<?php

namespace App\Http\Controllers;

use App\Chat\ChatError;
use App\Chat\Envelope;
use App\Chat\Rooms;
use App\Events\ChatEnvelope;
use App\Models\User;
use Illuminate\Database\UniqueConstraintViolationException;
use Illuminate\Http\JsonResponse;
use Illuminate\Http\Request;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\RateLimiter;

class ChatController extends Controller
{
    public function __invoke(Request $request): JsonResponse
    {
        $env = $request->json()->all();
        try {
            return response()->json($this->handle($request->user(), $env));
        } catch (ChatError $e) {
            $error = Envelope::make('error', null, $env['room'] ?? null, [
                'ref' => is_string($env['id'] ?? null) ? $env['id'] : null,
                'code' => $e->codeName, 'message' => $e->getMessage(), 'retryable' => $e->retryable(),
            ]);

            return response()->json($error, $e->status());
        }
    }

    private function handle(User $user, array $env): array
    {
        // rate limit first: 20 envelopes per second per user, checked before deduplication
        if (! RateLimiter::attempt('chat:'.$user->id, 20, fn () => true, 1)) {
            throw new ChatError('rate_limited', 'Slow down.');
        }
        if (($env['v'] ?? null) !== 1) {
            throw new ChatError('unsupported_version', 'Only protocol version 1 is supported.');
        }
        $type = $env['type'] ?? null;
        $id = $env['id'] ?? null;
        $room = $env['room'] ?? null;
        $payload = $env['payload'] ?? null;
        if (! in_array($type, ['message', 'edit', 'react', 'history'], true)) {
            throw new ChatError('bad_request', 'This endpoint carries message, edit, react and history.');
        }
        if (! is_string($id) || ! preg_match('/^[0-9A-Za-z-]{26,36}$/', $id)) {
            throw new ChatError('bad_request', 'id must be 26 to 36 characters.');
        }
        if (! is_string($room) || ! Rooms::valid($room) || ! is_array($payload)) {
            throw new ChatError('bad_request', 'room or payload is invalid.');
        }
        if (! Rooms::canAccess($user, $room)) {
            throw new ChatError('forbidden', 'You cannot use this room.');
        }
        if ($type === 'history') {
            return $this->history($id, $room, $payload);
        }
        if ($stored = DB::table('chat_acks')->where(['user_id' => $user->id, 'id' => $id])->value('ack')) {
            $ack = json_decode($stored);                 // the original ack, flagged as a duplicate
            $ack->payload->dup = true;

            return (array) $ack;
        }

        DB::table('chat_rooms')->insertOrIgnore(['name' => $room, 'last_seq' => 0]);   // outside the transaction: no gap locks
        $broadcast = null;
        try {
            $ack = DB::transaction(function () use ($user, $type, $id, $room, $payload, &$broadcast) {
                $seq = $this->nextSeq($room);
                $broadcast = match ($type) {
                    'message' => $this->message($user, $id, $room, $payload, $seq),
                    'edit' => $this->edit($user, $room, $payload, $seq),
                    'react' => $this->react($user, $room, $payload, $seq),
                };
                $ack = Envelope::ack($id, $room, $seq);
                DB::table('chat_acks')->insert([
                    'user_id' => $user->id, 'id' => $id, 'ack' => json_encode($ack), 'created_ms' => Envelope::nowMs(),
                ]);

                return $ack;
            }, attempts: 3);
        } catch (UniqueConstraintViolationException) {
            throw new ChatError('conflict', 'That id or reaction already exists.');
        }
        ChatEnvelope::dispatch($room, $broadcast);   // after the commit, so nobody sees a rolled-back seq

        return $ack;
    }

    private function nextSeq(string $room): int
    {
        // the row exists (see handle()); lockForUpdate serialises senders in a room until their commit
        $seq = (int) DB::table('chat_rooms')->where('name', $room)->lockForUpdate()->value('last_seq') + 1;
        DB::table('chat_rooms')->where('name', $room)->update(['last_seq' => $seq]);

        return $seq;
    }

    private function message(User $user, string $id, string $room, array $p, int $seq): array
    {
        $text = $p['text'] ?? null;
        $reply = $p['replyTo'] ?? null;
        $attachments = $p['attachments'] ?? [];
        if (! is_string($text) || ($len = strlen(mb_convert_encoding($text, 'UTF-16LE', 'UTF-8')) / 2) < 1 || $len > 4000) {
            throw new ChatError('bad_request', 'text must be 1 to 4000 UTF-16 code units.');
        }
        if (! is_array($attachments) || count($attachments) > 5) {
            throw new ChatError('bad_request', 'At most 5 attachments.');
        }
        if ($reply !== null && ! DB::table('chat_messages')->where(['id' => $reply, 'room' => $room])->exists()) {
            throw new ChatError('not_found', 'replyTo is not a message of this room.');
        }
        $now = Envelope::nowMs();
        DB::table('chat_messages')->insert([
            'id' => $id, 'room' => $room, 'user_id' => $user->id, 'seq' => $seq, 'text' => $text,
            'reply_to' => $reply, 'attachments' => json_encode($attachments), 'created_ms' => $now,
        ]);

        return Envelope::make('message', $id, $room, [
            'from' => ['userId' => (string) $user->id, 'name' => $user->name], 'text' => $text,
            'replyTo' => $reply, 'attachments' => $attachments, 'seq' => $seq, 'edited' => false,
        ], $now);
    }

    private function edit(User $user, string $room, array $p, int $seq): array
    {
        $row = DB::table('chat_messages')->where(['id' => $p['target'] ?? '', 'room' => $room])->first();
        if (! $row) {
            throw new ChatError('not_found', 'Unknown target.');
        }
        if ($row->user_id !== $user->id) {
            throw new ChatError('forbidden', 'Only the author can edit.');
        }
        if (! is_string($p['text'] ?? null) || $p['text'] === '') {
            throw new ChatError('bad_request', 'text is required.');
        }
        $now = Envelope::nowMs();
        DB::table('chat_messages')->where('id', $row->id)->update(['text' => $p['text'], 'edited_ms' => $now]);

        return Envelope::make('edit', null, $room, ['target' => $row->id, 'text' => $p['text'], 'seq' => $seq, 'editedAt' => $now], $now);
    }

    private function react(User $user, string $room, array $p, int $seq): array
    {
        $emoji = $p['emoji'] ?? '';
        $op = $p['op'] ?? '';
        if (! DB::table('chat_messages')->where(['id' => $p['target'] ?? '', 'room' => $room])->exists()) {
            throw new ChatError('not_found', 'Unknown target.');
        }
        if (! is_string($emoji) || preg_match_all('/\X/u', $emoji) !== 1 || ! in_array($op, ['add', 'remove'], true)) {
            throw new ChatError('bad_request', 'One emoji and op add or remove.');
        }
        $key = ['target' => $p['target'], 'user_id' => $user->id, 'emoji' => $emoji];
        if ($op === 'add') {
            DB::table('chat_reactions')->insert($key);          // a second add violates the primary key: conflict
        } elseif (DB::table('chat_reactions')->where($key)->delete() === 0) {
            throw new ChatError('not_found', 'You have not added that emoji.');
        }

        return Envelope::make('react', null, $room, [
            'target' => $p['target'], 'emoji' => $emoji, 'op' => $op, 'from' => ['userId' => (string) $user->id], 'seq' => $seq,
        ]);
    }

    private function history(string $id, string $room, array $p): array
    {
        $before = $p['before'] ?? null;
        $limit = $p['limit'] ?? 50;
        if (($before !== null && ! is_int($before)) || ! is_int($limit) || $limit < 1 || $limit > 100) {
            throw new ChatError('bad_request', 'before is a seq or null, limit is 1 to 100.');
        }
        $rows = DB::table('chat_messages')->join('users', 'users.id', '=', 'chat_messages.user_id')
            ->where('room', $room)->when($before !== null, fn ($q) => $q->where('seq', '<', $before))
            ->orderByDesc('seq')->limit($limit + 1)
            ->get(['chat_messages.*', 'users.name'])->all();
        $hasMore = count($rows) > $limit;
        $rows = array_reverse(array_slice($rows, 0, $limit));       // ascending seq
        $messages = array_map(fn ($r) => Envelope::make('message', $r->id, $room, [
            'from' => ['userId' => (string) $r->user_id, 'name' => $r->name], 'text' => $r->text,
            'replyTo' => $r->reply_to, 'attachments' => json_decode($r->attachments, true),
            'seq' => $r->seq, 'edited' => $r->edited_ms !== null,
        ], $r->created_ms), $rows);

        return Envelope::ack($id, $room, null, [
            'messages' => $messages, 'hasMore' => $hasMore, 'next' => $hasMore ? $rows[0]->seq : null,
        ]);
    }
}

// resources/js/chatterbox/Room.tsx
import { configureEcho, useConnectionStatus, useEchoPresence } from '@laravel/echo-react';
import { useCallback, useEffect, useRef, useState } from 'react';

configureEcho({
  broadcaster: 'reverb',
  key: import.meta.env.VITE_REVERB_APP_KEY,
  wsHost: import.meta.env.VITE_REVERB_HOST,
  wsPort: Number(import.meta.env.VITE_REVERB_PORT ?? 80),
  wssPort: Number(import.meta.env.VITE_REVERB_PORT ?? 443),
  forceTLS: (import.meta.env.VITE_REVERB_SCHEME ?? 'https') === 'https',
  enabledTransports: ['ws', 'wss'],
  authEndpoint: '/api/broadcasting/auth',
  auth: { headers: { Authorization: `Bearer ${localStorage.getItem('chatterbox.token')}`, Accept: 'application/json' } },
});

type Message = { id: string; payload: { seq: number; text: string; edited: boolean; from: { userId: string; name: string } } };

async function call(type: string, room: string, payload: object, id = crypto.randomUUID()) {
  const response = await fetch('/api/chat', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json', Accept: 'application/json', Authorization: `Bearer ${localStorage.getItem('chatterbox.token')}` },
    body: JSON.stringify({ v: 1, type, id, room, ts: 0, payload }),
  });
  return { ok: response.ok, ...(await response.json()) };      // an ack or error envelope either way
}

export function Room({ room }: { room: string }) {
  const [messages, setMessages] = useState<Message[]>([]);
  const [typing, setTyping] = useState<string[]>([]);
  const seen = useRef(new Set<number>());
  const status = useConnectionStatus();                         // connected, connecting, reconnecting, disconnected, failed

  const merge = useCallback((incoming: Message[]) => setMessages((all) => {
    const fresh = incoming.filter((m) => !seen.current.has(m.payload.seq) && seen.current.add(m.payload.seq));
    return [...all, ...fresh].sort((a, b) => a.payload.seq - b.payload.seq);   // seq is the ordering key
  }), []);

  const { channel } = useEchoPresence<Message>(`chat.${room}`, '.message', (m) => merge([m]));   // leading dot: no namespace

  useEffect(() => {
    if (status !== 'connected') return;
    // subscribe first, then fetch, so nothing falls into the gap; duplicates are dropped by seq
    call('history', room, { before: null, limit: 50 }).then((ack) => merge(ack.payload.result.messages));
    const presence = channel();
    presence?.listenForWhisper('typing', (e: { userId: string; state: string }) =>
      setTyping((names) => (e.state === 'start' ? [...new Set([...names, e.userId])] : names.filter((n) => n !== e.userId))));
  }, [status, room]);

  const send = (text: string) => call('message', room, { text, replyTo: null, attachments: [] });
  return <p>{status}: {messages.length} messages, typing: {typing.join(', ')} <button onClick={() => send('Hello, Chatterbox')}>Send</button></p>;
}

// Real laravel-echo 2.6.1 + pusher-js 8.6.0 in Node. pusher-js sends no Origin header here, so the
// Reverb on port 8233 runs with REVERB_ALLOWED_ORIGINS='*' (a browser always sends its Origin).
import Echo from 'laravel-echo';
import Pusher from 'pusher-js';
import { readFileSync } from 'node:fs';
import { envelope, send } from './chatterbox-client.mjs';

const tokens = Object.fromEntries(readFileSync('tokens.txt', 'utf8').trim().split('\n').map((l) => l.split(' ').slice(1)));
const connect = (user) => new Echo({
  broadcaster: 'reverb', Pusher, key: process.env.REVERB_APP_KEY,
  wsHost: 'localhost', wsPort: 8233, forceTLS: false, enabledTransports: ['ws'],
  authEndpoint: 'http://localhost:8230/api/broadcasting/auth',
  auth: { headers: { Authorization: `Bearer ${tokens[user]}`, Accept: 'application/json' } },
});
const pause = (ms) => new Promise((r) => setTimeout(r, ms));

const bob = connect('bob');
bob.join('chat.general')
  .here((members) => console.log('here    ', JSON.stringify(members)))
  .joining((m) => console.log('joining ', JSON.stringify(m)))
  .leaving((m) => console.log('leaving ', JSON.stringify(m)))
  .listen('.message', (e) => console.log('message ', e.payload.seq, e.payload.text))
  .listenForWhisper('typing', (e) => console.log('typing  ', JSON.stringify(e)));
await pause(1500);

const alice = connect('alice');
const room = alice.join('chat.general');
await pause(1500);
room.whisper('typing', { state: 'start' });
console.log('ack     ', (await send('alice', envelope('message', 'general', { text: 'sent through Echo' }))).seq);
await pause(1000);
console.log('state   ', alice.connector.pusher.connection.state, typeof alice.socketId());
alice.leave('chat.general');
await pause(1000);
alice.disconnect(); bob.disconnect();
process.exit(0);

here     [{"userId":"2","name":"Bob"}]
joining  {"userId":"1","name":"Alice"}
typing   {"state":"start"}
message  1 sent through Echo
ack      1
state    connected string
leaving  {"userId":"1","name":"Alice"}

upstream reverb { server reverb1:8080; server reverb2:8080; }   # round robin; the Redis backplane makes any instance fine

map $http_upgrade $connection_upgrade { default upgrade; '' close; }

server {
    listen 80;
    server_name chat.example.test;

    location /app {                          # WebSocket (Pusher protocol)
        proxy_pass http://reverb;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection $connection_upgrade;
        proxy_set_header Host $http_host;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_read_timeout 120s;             # longer than the 60 s ping interval
    }
    location /apps { proxy_pass http://reverb; proxy_set_header Host $http_host; }   # HTTP events API
}

<?php

namespace Tests\Feature;

use App\Events\ChatEnvelope;
use App\Models\User;
use Illuminate\Foundation\Testing\RefreshDatabase;
use Illuminate\Support\Facades\Event;
use Tests\TestCase;

class ChatterboxTest extends TestCase
{
    use RefreshDatabase;

    private function envelope(string $type, string $room, array $payload, string $id = '01JAB3W7Q8Z4N5R6T7V8X9Y0ZA'): array
    {
        return ['v' => 1, 'type' => $type, 'id' => $id, 'room' => $room, 'ts' => 0, 'payload' => $payload];
    }

    public function test_only_the_two_users_of_a_dm_can_authorize_its_presence_channel(): void
    {
        [$a, $b, $c] = User::factory()->count(3)->create();
        $channel = "presence-chat.dm-{$a->id}-{$b->id}";

        foreach ([$a, $b] as $member) {
            $this->actingAs($member, 'sanctum')
                ->postJson('/api/broadcasting/auth', ['socket_id' => '1234.5678', 'channel_name' => $channel])
                ->assertOk()->assertJsonStructure(['auth', 'channel_data']);
        }
        $this->actingAs($c, 'sanctum')
            ->postJson('/api/broadcasting/auth', ['socket_id' => '1234.5678', 'channel_name' => $channel])
            ->assertForbidden();
    }

    public function test_a_message_gets_a_seq_is_broadcast_once_and_a_retry_is_a_duplicate(): void
    {
        Event::fake([ChatEnvelope::class]);
        $this->actingAs(User::factory()->create(), 'sanctum');
        $message = $this->envelope('message', 'general', ['text' => 'Hello, Chatterbox', 'replyTo' => null, 'attachments' => []]);

        $first = $this->postJson('/api/chat', $message)->assertOk();
        $retry = $this->postJson('/api/chat', $message)->assertOk();

        $this->assertSame(1, $first->json('payload.seq'));
        $this->assertFalse($first->json('payload.dup'));
        $this->assertTrue($retry->json('payload.dup'));
        $this->assertSame(1, $retry->json('payload.seq'));
        Event::assertDispatchedTimes(ChatEnvelope::class, 1);
        Event::assertDispatched(ChatEnvelope::class, fn ($e) => $e->room === 'general' && $e->envelope['payload']->seq === 1);
    }

    public function test_errors_are_envelopes_with_the_chatterbox_codes(): void
    {
        $this->actingAs(User::factory()->create(), 'sanctum');

        $this->postJson('/api/chat', $this->envelope('message', 'dm-998-999', ['text' => 'hi']))
            ->assertForbidden()->assertJsonPath('type', 'error')->assertJsonPath('payload.code', 'forbidden');
        $this->postJson('/api/chat', ['v' => 2] + $this->envelope('message', 'general', ['text' => 'hi']))
            ->assertStatus(426)->assertJsonPath('payload.code', 'unsupported_version');
        $this->postJson('/api/chat', $this->envelope('message', 'general', ['text' => 'hi', 'replyTo' => 'missing']))
            ->assertNotFound()->assertJsonPath('payload.retryable', false);
    }
}

<env name="BROADCAST_CONNECTION" value="reverb"/>
<env name="REVERB_APP_ID" value="test-app"/>
<env name="REVERB_APP_KEY" value="test-key"/>
<env name="REVERB_APP_SECRET" value="test-secret"/>

// Minimal Pusher-protocol client for tests: Node 24 + ws 8.22.0, no Echo.
import WebSocket from 'ws';
import { readFileSync } from 'node:fs';

export const KEY = process.env.REVERB_APP_KEY;
export const API = 'http://localhost:8230/api';
const tokens = Object.fromEntries(readFileSync('tokens.txt', 'utf8').trim().split('\n').map((l) => l.split(' ').slice(1)));
const headers = (user) => ({ Authorization: `Bearer ${tokens[user]}`, 'Content-Type': 'application/json', Accept: 'application/json' });

let counter = 0;
const run = Math.random().toString(36).slice(2, 8).toUpperCase();
export const newId = () => `01JAB3W7Q8Z4N5R6${run}${1000 + counter++}`;                 // 26 characters
export const envelope = (type, room, payload, id = newId()) => ({ v: 1, type, id, room, ts: 0, payload });

export async function send(user, env) {                                                    // writes go over HTTP
  const res = await fetch(`${API}/chat`, { method: 'POST', headers: headers(user), body: JSON.stringify(env) });
  const body = await res.json();
  return { status: res.status, ...body.payload, type: body.type };
}

export class Socket {
  frames = [];
  constructor(user, port) {
    this.user = user;
    this.ws = new WebSocket(`ws://localhost:${port}/app/${KEY}?protocol=7&client=test&version=1`, { origin: 'https://chat.example.test' });
    this.ws.on('message', (raw) => {
      const frame = JSON.parse(raw);
      if (typeof frame.data === 'string') frame.data = JSON.parse(frame.data);          // data is double-encoded
      if (frame.event === 'pusher:connection_established') this.socketId = frame.data.socket_id;
      this.frames.push(frame);
    });
  }
  async until(test, ms = 3000) {
    for (const end = Date.now() + ms; Date.now() < end; await new Promise((r) => setTimeout(r, 20))) {
      const hit = this.frames.find(test);
      if (hit) return hit;
    }
    throw new Error('timeout');
  }
  async join(channel) {                                                                  // authorize over HTTP, then subscribe
    await this.until((f) => f.event === 'pusher:connection_established');
    const res = await fetch(`${API}/broadcasting/auth`, { method: 'POST', headers: headers(this.user), body: JSON.stringify({ socket_id: this.socketId, channel_name: channel }) });
    if (!res.ok) return { denied: res.status };
    this.ws.send(JSON.stringify({ event: 'pusher:subscribe', data: { channel, ...(await res.json()) } }));
    return this.until((f) => f.event === 'pusher_internal:subscription_succeeded' && f.channel === channel);
  }
  whisper(channel, event, data) { this.ws.send(JSON.stringify({ event: `client-${event}`, channel, data })); }
  count(event) { return this.frames.filter((f) => f.event === event).length; }
}

import assert from 'node:assert/strict';
import { Socket, envelope, send } from './chatterbox-client.mjs';

const room = 'presence-chat.general';
const alice = new Socket('alice', 8231);   // Reverb instance 1
const bob = new Socket('bob', 8232);       // Reverb instance 2
const mallory = new Socket('mallory', 8231);

const joined = await alice.join(room);
assert.deepEqual(Object.keys(joined.data.presence.hash), ['1']);
await bob.join(room);
await alice.until((f) => f.event === 'pusher_internal:member_added');                   // crossed instances, via Redis
assert.deepEqual(await mallory.join('presence-chat.dm-1-2'), { denied: 403 });          // not a member of the DM

const hello = envelope('message', 'general', { text: 'Hello, Chatterbox', replyTo: null, attachments: [] });
const ack = await send('alice', hello);
assert.deepEqual([ack.type, ack.ok, ack.dup, typeof ack.seq], ['ack', true, false, 'number']);
const pushed = await bob.until((f) => f.event === 'message');                           // bob is on the other instance
assert.equal(pushed.data.id, hello.id);
assert.equal(pushed.data.payload.seq, ack.seq);
await alice.until((f) => f.event === 'message');                                         // the sender gets its copy too

const retry = await send('alice', hello);                                                // same id again
assert.deepEqual([retry.dup, retry.seq], [true, ack.seq]);
await new Promise((r) => setTimeout(r, 300));
assert.equal(bob.count('message'), 1);                                                   // no second broadcast

const code = async (user, env) => (await send(user, env)).code;
assert.equal(await code('bob', envelope('edit', 'general', { target: hello.id, text: 'x' })), 'forbidden');
assert.equal(await code('alice', envelope('message', 'general', { text: 'hi', replyTo: 'missing' })), 'not_found');
assert.equal(await code('alice', { ...envelope('message', 'general', { text: 'hi' }), v: 2 }), 'unsupported_version');
assert.equal(await code('alice', envelope('message', 'general', { text: '' })), 'bad_request');
assert.equal(await code('mallory', envelope('message', 'dm-1-2', { text: 'hi' })), 'forbidden');
const like = (op) => envelope('react', 'general', { target: hello.id, emoji: '👍', op });
assert.equal((await send('bob', like('add'))).ok, true);
assert.equal(await code('bob', like('add')), 'conflict');

alice.whisper(room, 'typing', { state: 'start' });                                       // peer to peer, never touches Laravel
assert.deepEqual((await bob.until((f) => f.event === 'client-typing')).data, { state: 'start' });
assert.equal(alice.count('client-typing'), 0);                                           // not echoed to the sender

const page = (await send('bob', envelope('history', 'general', { before: null, limit: 2 })));
assert.deepEqual([page.result.messages.length, page.result.hasMore], [1, false]);       // edit and react consumed seq 2 and 3
console.log('all Chatterbox-on-Reverb checks passed');
process.exit(0);