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
| 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 |
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')),
],
],
.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
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
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);
}
}
