WebSocket (Jetstream / Firehose)

Funzionalità WebSocket di Laravel Bluesky. Come gestire eventi in tempo reale con Bluesky Jetstream e AT Protocol Firehose. Spiega l’uso di JetstreamServeCommand e FirehoseServeCommand.

​

Panoramica

laravel-bluesky offre due comandi WebSocket per collegarsi agli stream in tempo reale di Bluesky.

• Jetstream — endpoint WebSocket con filtri, proprio di Bluesky. Leggero, in formato JSON.
• Firehose — stream grezzo di eventi di AT Protocol. Riceve tutti i dati in formato binario DAG-CBOR.

I processi long-running via WebSocket richiedono un server sempre attivo, come VPS o EC2, oppure un custom worker di Laravel Cloud. Non funzionano in ambienti serverless come Laravel Vapor o Vercel.

​

Installazione

Per le funzionalità WebSocket è necessario Workerman.

composer require workerman/workerman

​

Jetstream

​

Panoramica

Jetstream è il servizio WebSocket con filtraggio fornito da Bluesky. Puoi filtrare per tipo di collezione e DID utente, ricevendo solo gli eventi necessari in modo efficiente.

Caratteristica	Descrizione
Formato dati	JSON
Filtraggio	Possibile per collezione e DID
Volume dati	Leggero a seconda del filtro
Uso	Monitoraggio di post, like, follow, ecc.

​

Come avviarlo

# Ricevi tutti i messaggi (nessun filtro)
php artisan bluesky:ws start

# Debug: mostra tutti i messaggi ricevuti
php artisan bluesky:ws start -v

​

Filtro per collezione

Con l’opzione -C restringi le collezioni ricevute. Puoi specificarne più di una.

# Ricevi solo post e like
php artisan bluesky:ws start -C app.bsky.feed.post -C app.bsky.feed.like

# Ricevi solo i follow
php artisan bluesky:ws start -C app.bsky.graph.follow

Collezioni principali:

Collezione	Contenuto
app.bsky.feed.post	Creazione/eliminazione di post
app.bsky.feed.like	Like
app.bsky.feed.repost	Repost
app.bsky.graph.follow	Follow
app.bsky.graph.block	Block

​

Filtro per DID

Con l’opzione -D ricevi solo gli eventi di utenti specifici.

# Ricevi solo i post di determinati utenti
php artisan bluesky:ws start -C app.bsky.feed.post -D did:plc:xxx -D did:plc:yyy

​

Gestione degli eventi

Il comando Jetstream emette eventi Laravel in base al tipo di messaggio ricevuto.

Classe evento	Quando
JetstreamMessageReceived	A ogni messaggio ricevuto
JetstreamCommitMessage	Alla creazione/aggiornamento/eliminazione di un record
JetstreamIdentityMessage	Su eventi di identity, come cambio handle
JetstreamAccountMessage	All’attivazione/disattivazione dell’account

Crea un listener per gestire gli eventi.

php artisan make:listener JetstreamPostListener

namespace App\Listeners;

use Revolution\Bluesky\Events\Jetstream\JetstreamCommitMessage;

class JetstreamPostListener
{
    public function handle(JetstreamCommitMessage $event): void
    {
        // Verifica il tipo di collezione
        $collection = data_get($event->message, 'commit.collection');

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

        // Tipo di operazione: create / update / delete
        $operation = $event->operation;

        // DID dell'autore del post
        $did = $event->message['did'];

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

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

​

Firehose

​

Panoramica

Firehose è lo stream grezzo di eventi di AT Protocol. Ricevi in formato binario (DAG-CBOR) tutte le operazioni sui record della rete Bluesky.

Caratteristica	Descrizione
Formato dati	Binario DAG-CBOR (il pacchetto lo decodifica in automatico)
Filtraggio	Nessuno (ricevi tutti i dati)
Volume dati	Molto elevato
Uso	Raccolta e archiviazione di tutti i dati

La decodifica DAG-CBOR viene fatta automaticamente dal pacchetto. Negli event listener ricevi i dati come normali array PHP.

​

Come avviarlo

php artisan bluesky:firehose start

# Debug: mostra i messaggi ricevuti
php artisan bluesky:firehose start -v

​

Gestione degli eventi

Anche il comando Firehose gestisce i messaggi tramite eventi Laravel.

Classe evento	Quando
FirehoseMessageReceived	A ogni messaggio ricevuto (dati grezzi inclusi)
FirehoseCommitMessage	Alla creazione/aggiornamento/eliminazione di un record
FirehoseIdentityMessage	Su eventi di identity
FirehoseAccountMessage	Su eventi di account
FirehoseSyncMessage	Su eventi di sincronizzazione del repository

php artisan make:listener FirehosePostListener

namespace App\Listeners;

use Revolution\Bluesky\Events\Firehose\FirehoseCommitMessage;

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

        // Tipo di operazione: create / update / delete
        $action = $event->action;

        // DID dell'autore del post
        $did = $event->did;

        // Contenuto del record (array decodificato)
        $record = $event->record;
        $text = data_get($record, 'value.text', '');

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

​

Configurazione

In config/bluesky.php puoi modificare l’host di connessione e le impostazioni di logging.

// Jetstream
'jetstream' => [
    'host' => env('BLUESKY_JETSTREAM_HOST', 'jetstream1.us-west.bsky.network'),
    'max' => env('BLUESKY_JETSTREAM_MAX', 0), // maxMessageSizeBytes (0 = senza limite)
    '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')),
    ],
],

Esempio di configurazione in .env:

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

​

Combinazione con Labeler

Puoi avviare contemporaneamente il server Labeler e Jetstream / Firehose. Puoi usare i dati di Jetstream o Firehose per gestire le richieste di labeling ricevute dal Labeler.

# Labeler + Jetstream (monitora gli eventi follow)
php artisan bluesky:labeler:server start --jetstream -C app.bsky.graph.follow

# Labeler + Firehose (ricevi tutti i dati)
php artisan bluesky:labeler:server start --firehose

Per i dettagli su Labeler consulta la pagina Labeler.

​

Gestione dei processi long-running

I comandi WebSocket sono processi che restano attivi a lungo. In produzione usa uno strumento di gestione dei processi come Supervisor.

​

Esempio di configurazione 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

# Ricarica Supervisor e avvia
sudo supervisorctl reread
sudo supervisorctl update
sudo supervisorctl start bluesky-jetstream:*

​

Configurazione del daemon su Laravel Forge

Se usi Laravel Forge, aggiungi il daemon dalla sezione Daemons.

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

​

Configurazione dei processi in background su Laravel Cloud

I comandi WebSocket si connettono allo stream di Bluesky come client WebSocket, quindi funzionano anche su Laravel Cloud. Configurali come processi in background (custom worker) di Laravel Cloud.

Nelle impostazioni dei processi in background di Laravel Cloud, aggiungi un Custom Worker.

Per Jetstream:

php artisan bluesky:ws start

Per Firehose:

php artisan bluesky:firehose start

L’arresto e il riavvio del processo al deploy sono gestiti automaticamente da Laravel Cloud. A parte la configurazione del processo in background non serve nient’altro.

​

Attenzione

• Se il processo termina inaspettatamente, con autorestart=true si riavvia automaticamente.
• Per evitare memory leak, valuta riavvii periodici.
• Con Firehose, che riceve grandi quantità di messaggi, si consiglia di eseguire il lavoro nel listener in modo asincrono (Queue Job).

// Esempio di dispatch verso un Queue Job dal 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 l'elaborazione pesante a un Queue Job
        ProcessFirehosePost::dispatch($event->did, $event->record, $event->action);
    }
}

Source: src/Console/WebSocket