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