WebSocket (Jetstream / Firehose)

Funcionalidad WebSocket de Laravel Bluesky. Cómo procesar eventos en tiempo real usando Bluesky Jetstream y el Firehose de AT Protocol. Se explica el uso de JetstreamServeCommand y FirehoseServeCommand.

​

Descripción general

laravel-bluesky proporciona dos comandos WebSocket para conectarse a los streams en tiempo real de Bluesky.

• Jetstream: endpoint WebSocket con filtrado propio de Bluesky. Ligero, en formato JSON.
• Firehose: stream de eventos crudo de AT Protocol. Recibe todos los datos en formato binario DAG-CBOR.

Los procesos WebSocket de larga duración requieren un servidor siempre activo como VPS o EC2, o un worker personalizado en Laravel Cloud. No funcionan en entornos serverless como Laravel Vapor o Vercel.

​

Instalación

La funcionalidad WebSocket requiere Workerman.

composer require workerman/workerman

​

Jetstream

​

Descripción general

Jetstream es un servicio WebSocket con filtrado proporcionado por Bluesky. Como se puede filtrar por tipo de colección o por DID de usuario, puedes recibir eficientemente solo los eventos que necesitas.

Característica	Descripción
Formato de datos	JSON
Filtrado	Filtrable por colección o DID
Volumen de datos	Ligero según el filtro
Uso	Monitorizar publicaciones, likes, follows, etc.

​

Cómo iniciarlo

# Recibe todos los mensajes (sin filtro)
php artisan bluesky:ws start

# Depuración: muestra todos los mensajes recibidos
php artisan bluesky:ws start -v

​

Filtro por colección

Con la opción -C puedes restringir las colecciones que quieres recibir. Se pueden indicar varias.

# Recibe solo publicaciones y likes
php artisan bluesky:ws start -C app.bsky.feed.post -C app.bsky.feed.like

# Recibe solo follows
php artisan bluesky:ws start -C app.bsky.graph.follow

Principales colecciones:

Colección	Contenido
app.bsky.feed.post	Creación y eliminación de publicaciones
app.bsky.feed.like	Likes
app.bsky.feed.repost	Reposts
app.bsky.graph.follow	Follows
app.bsky.graph.block	Bloqueos

​

Filtro por DID

Con la opción -D recibes solo los eventos de usuarios concretos.

# Recibe solo publicaciones de determinados usuarios
php artisan bluesky:ws start -C app.bsky.feed.post -D did:plc:xxx -D did:plc:yyy

​

Procesamiento de eventos

El comando Jetstream dispara eventos de Laravel según el tipo de mensaje recibido.

Clase de evento	Momento
JetstreamMessageReceived	Al recibir cualquier mensaje
JetstreamCommitMessage	Al crear, actualizar o eliminar un record
JetstreamIdentityMessage	En eventos identity como cambio de handle
JetstreamAccountMessage	Al activar o desactivar cuentas

Crea un listener para procesar los eventos.

php artisan make:listener JetstreamPostListener

namespace App\Listeners;

use Revolution\Bluesky\Events\Jetstream\JetstreamCommitMessage;

class JetstreamPostListener
{
    public function handle(JetstreamCommitMessage $event): void
    {
        // Comprueba el tipo de colección
        $collection = data_get($event->message, 'commit.collection');

        if ($collection !== 'app.bsky.feed.post') {
            return;
        }

        // Tipo de operación: create / update / delete
        $operation = $event->operation;

        // DID del autor
        $did = $event->message['did'];

        // Contenido del record
        $record = data_get($event->message, 'commit.record');
        $text = data_get($record, 'text', '');

        info("[$operation] $did: $text");
    }
}

​

Firehose

​

Descripción general

Firehose es el stream de eventos crudo de AT Protocol. Recibe todas las operaciones de record de la red Bluesky en formato binario (DAG-CBOR).

Característica	Descripción
Formato de datos	Binario DAG-CBOR (el paquete lo decodifica automáticamente)
Filtrado	Ninguno (recibe todos los datos)
Volumen de datos	Muy grande
Uso	Recogida y archivado de todos los datos

La decodificación DAG-CBOR la realiza el paquete automáticamente. En los listeners recibes los datos como arrays PHP normales.

​

Cómo iniciarlo

php artisan bluesky:firehose start

# Depuración: muestra los mensajes recibidos
php artisan bluesky:firehose start -v

​

Procesamiento de eventos

El comando Firehose también usa eventos de Laravel para procesar los mensajes.

Clase de evento	Momento
FirehoseMessageReceived	Al recibir cualquier mensaje (incluye datos crudos)
FirehoseCommitMessage	Al crear, actualizar o eliminar un record
FirehoseIdentityMessage	En eventos identity
FirehoseAccountMessage	En eventos de cuenta
FirehoseSyncMessage	En eventos de sincronización del repositorio

php artisan make:listener FirehosePostListener

namespace App\Listeners;

use Revolution\Bluesky\Events\Firehose\FirehoseCommitMessage;

class FirehosePostListener
{
    public function handle(FirehoseCommitMessage $event): void
    {
        // Comprueba el tipo de colección
        if ($event->collection !== 'app.bsky.feed.post') {
            return;
        }

        // Tipo de operación: create / update / delete
        $action = $event->action;

        // DID del autor
        $did = $event->did;

        // Contenido del record (array ya decodificado)
        $record = $event->record;
        $text = data_get($record, 'value.text', '');

        info("[$action] $did: $text");
    }
}

​

Configuración

En config/bluesky.php puedes cambiar el host de conexión y la configuración de logs.

// Jetstream
'jetstream' => [
    'host' => env('BLUESKY_JETSTREAM_HOST', 'jetstream1.us-west.bsky.network'),
    'max' => env('BLUESKY_JETSTREAM_MAX', 0), // maxMessageSizeBytes (0 = sin límite)
    'logging' => [
        'driver' => env('BLUESKY_JETSTREAM_LOG_DRIVER', 'daily'),
        'days' => 7,
        'path' => env('BLUESKY_JETSTREAM_LOG_PATH', storage_path('logs/jetstream.log')),
    ],
],

// Firehose
'firehose' => [
    'host' => env('BLUESKY_FIREHOSE_HOST', 'bsky.network'),
    'logging' => [
        'driver' => env('BLUESKY_FIREHOSE_LOG_DRIVER', 'daily'),
        'days' => 7,
        'path' => env('BLUESKY_FIREHOSE_LOG_PATH', storage_path('logs/firehose.log')),
    ],
],

Ejemplo de configuración en .env:

BLUESKY_JETSTREAM_HOST=jetstream2.us-east.bsky.network
BLUESKY_JETSTREAM_MAX=1000000

​

Combinación con Labeler

Puedes iniciar el servidor Labeler junto con Jetstream / Firehose. El Labeler puede aprovechar los datos de Jetstream o Firehose para procesar las solicitudes de etiquetado que recibe.

# Labeler + Jetstream (monitoriza eventos de follow)
php artisan bluesky:labeler:server start --jetstream -C app.bsky.graph.follow

# Labeler + Firehose (recibe todos los datos)
php artisan bluesky:labeler:server start --firehose

Para más detalles sobre Labeler, consulta la página Labeler.

​

Operación de procesos de larga duración

Los comandos WebSocket son procesos que permanecen activos durante mucho tiempo. En producción, usa una herramienta de gestión de procesos como Supervisor.

​

Ejemplo de configuración de Supervisor

/etc/supervisor/conf.d/bluesky-jetstream.conf:

[program:bluesky-jetstream]
process_name=%(program_name)s_%(process_num)02d
command=php /var/www/html/artisan bluesky:ws start -C app.bsky.feed.post
autostart=true
autorestart=true
stopasgroup=true
killasgroup=true
user=www-data
numprocs=1
redirect_stderr=true
stdout_logfile=/var/www/html/storage/logs/jetstream-worker.log
stopwaitsecs=3600

# Recarga y arranca Supervisor
sudo supervisorctl reread
sudo supervisorctl update
sudo supervisorctl start bluesky-jetstream:*

​

Configuración de daemons en Laravel Forge

Si usas Laravel Forge, añade el daemon desde la sección Daemons.

• Command: php artisan bluesky:ws start -C app.bsky.feed.post
• Directory: /var/www/html
• User: forge

​

Configuración de procesos en segundo plano en Laravel Cloud

Los comandos WebSocket actúan como cliente WebSocket que se conecta al stream de Bluesky, por lo que también funcionan en Laravel Cloud. Configúralos como proceso en segundo plano (custom worker) de Laravel Cloud.

En la configuración de procesos en segundo plano de Laravel Cloud, añade un Custom Worker.

Para Jetstream:

php artisan bluesky:ws start

Para Firehose:

php artisan bluesky:firehose start

Laravel Cloud gestiona automáticamente la parada y el reinicio del proceso durante el despliegue. Aparte de la configuración del proceso en segundo plano, no hace falta configuración adicional.

​

Consideraciones

• Si el proceso termina inesperadamente, se reiniciará automáticamente con autorestart=true.
• Para evitar fugas de memoria, considera reiniciarlo periódicamente.
• En Firehose, que recibe muchos mensajes, se recomienda que el procesamiento dentro del listener sea asíncrono (Queue Job).

// Ejemplo de despacho a Queue Job desde el listener

namespace App\Listeners;

use App\Jobs\ProcessFirehosePost;
use Revolution\Bluesky\Events\Firehose\FirehoseCommitMessage;

class FirehosePostListener
{
    public function handle(FirehoseCommitMessage $event): void
    {
        if ($event->collection !== 'app.bsky.feed.post') {
            return;
        }

        // Delega el procesamiento pesado a un Queue Job
        ProcessFirehosePost::dispatch($event->did, $event->record, $event->action);
    }
}

Source: src/Console/WebSocket