WebSocket (Jetstream / Firehose)

composer require workerman/workerman

# 接收所有訊息（不過濾）
php artisan bluesky:ws start

# 除錯：顯示所有接收到的訊息
php artisan bluesky:ws start -v

# 僅接收貼文與按讚
php artisan bluesky:ws start -C app.bsky.feed.post -C app.bsky.feed.like

# 僅接收追蹤
php artisan bluesky:ws start -C app.bsky.graph.follow

# 僅接收特定使用者的貼文
php artisan bluesky:ws start -C app.bsky.feed.post -D did:plc:xxx -D did:plc:yyy

php artisan make:listener JetstreamPostListener

namespace App\Listeners;

use Revolution\Bluesky\Events\Jetstream\JetstreamCommitMessage;

class JetstreamPostListener
{
    public function handle(JetstreamCommitMessage $event): void
    {
        // 確認 collection 種別
        $collection = data_get($event->message, 'commit.collection');

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

        // 操作類別：create / update / delete
        $operation = $event->operation;

        // 貼文者的 DID
        $did = $event->message['did'];

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

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

php artisan bluesky:firehose start

# 除錯：顯示接收到的訊息
php artisan bluesky:firehose start -v

php artisan make:listener FirehosePostListener

namespace App\Listeners;

use Revolution\Bluesky\Events\Firehose\FirehoseCommitMessage;

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

        // 操作類別：create / update / delete
        $action = $event->action;

        // 貼文者的 DID
        $did = $event->did;

        // record 內容（已解碼的陣列）
        $record = $event->record;
        $text = data_get($record, 'value.text', '');

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

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

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

# Labeler + Jetstream（監控追蹤事件）
php artisan bluesky:labeler:server start --jetstream -C app.bsky.graph.follow

# Labeler + Firehose（接收全部資料）
php artisan bluesky:labeler:server start --firehose

[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

# 重新載入 Supervisor 並啟動
sudo supervisorctl reread
sudo supervisorctl update
sudo supervisorctl start bluesky-jetstream:*

php artisan bluesky:ws start

php artisan bluesky:firehose start

// 於 listener 中派送至 Queue Job 的範例

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;
        }

        // 較重的處理交由 Queue Job
        ProcessFirehosePost::dispatch($event->did, $event->record, $event->action);
    }
}