WebSocket (Jetstream / Firehose)

Laravel Bluesky의 WebSocket 기능. Bluesky Jetstream과 AT Protocol Firehose를 사용한 실시간 이벤트 처리 방법. JetstreamServeCommand와 FirehoseServeCommand의 사용법을 설명합니다.

​

개요

laravel-bluesky는 Bluesky의 실시간 스트림에 접속하기 위한 2가지 WebSocket 명령어를 제공합니다.

• Jetstream — Bluesky 고유의 필터링된 WebSocket 엔드포인트. JSON 형식으로 경량.
• Firehose — AT Protocol의 원본 이벤트 스트림. DAG-CBOR 바이너리 형식으로 모든 데이터를 수신.

WebSocket에 의한 장시간 실행 프로세스는 VPS나 EC2 등의 상시 기동 서버 또는 Laravel Cloud의 커스텀 워커가 필요합니다. Laravel Vapor나 Vercel 등의 서버리스 환경에서는 동작하지 않습니다.

​

설치

WebSocket 기능에는 Workerman이 필요합니다.

composer require workerman/workerman

​

Jetstream

​

개요

Jetstream은 Bluesky가 제공하는 필터링된 WebSocket 서비스입니다. 컬렉션 종류나 사용자 DID로 필터링할 수 있어, 필요한 이벤트만 효율적으로 수신할 수 있습니다.

특징	설명
데이터 형식	JSON
필터링	컬렉션·DID로 필터 가능
데이터량	필터에 따라 경량
용도	게시·좋아요·팔로우 등 감시

​

기동 방법

# 모든 메시지 수신(필터 없음)
php artisan bluesky:ws start

# 디버그: 수신한 모든 메시지를 표시
php artisan bluesky:ws start -v

​

컬렉션 필터

-C 옵션으로 수신할 컬렉션을 좁힙니다. 여러 개 지정할 수 있습니다.

# 게시와 좋아요만 수신
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

주요 컬렉션:

컬렉션	내용
app.bsky.feed.post	게시 작성·삭제
app.bsky.feed.like	좋아요
app.bsky.feed.repost	리포스트
app.bsky.graph.follow	팔로우
app.bsky.graph.block	블록

​

DID 필터

-D 옵션으로 특정 사용자의 이벤트만 수신합니다.

# 특정 사용자의 게시만 수신
php artisan bluesky:ws start -C app.bsky.feed.post -D did:plc:xxx -D did:plc:yyy

​

이벤트 처리

Jetstream 명령어는 수신한 메시지의 종류에 따라 Laravel 이벤트를 발화합니다.

이벤트 클래스	타이밍
JetstreamMessageReceived	모든 메시지 수신 시
JetstreamCommitMessage	레코드 작성·갱신·삭제 시
JetstreamIdentityMessage	핸들 변경 등 identity 이벤트 시
JetstreamAccountMessage	계정 활성화·비활성화 시

이벤트 리스너를 만들어 이벤트를 처리합니다.

php artisan make:listener JetstreamPostListener

namespace App\Listeners;

use Revolution\Bluesky\Events\Jetstream\JetstreamCommitMessage;

class JetstreamPostListener
{
    public function handle(JetstreamCommitMessage $event): void
    {
        // 컬렉션 종류 확인
        $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 = data_get($event->message, 'commit.record');
        $text = data_get($record, 'text', '');

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

​

Firehose

​

개요

Firehose는 AT Protocol의 원본 이벤트 스트림입니다. Bluesky 네트워크상의 모든 레코드 조작을 바이너리(DAG-CBOR) 형식으로 수신합니다.

특징	설명
데이터 형식	DAG-CBOR 바이너리(패키지가 자동 디코드)
필터링	없음(모든 데이터 수신)
데이터량	매우 대량
용도	모든 데이터 수집·아카이브

DAG-CBOR 디코드는 패키지가 자동으로 수행합니다. 이벤트 리스너에서는 일반 PHP 배열로 데이터를 받을 수 있습니다.

​

기동 방법

php artisan bluesky:firehose start

# 디버그: 수신한 메시지 표시
php artisan bluesky:firehose start -v

​

이벤트 처리

Firehose 명령어도 Laravel 이벤트를 사용해 메시지를 처리합니다.

이벤트 클래스	타이밍
FirehoseMessageReceived	모든 메시지 수신 시(원본 데이터 포함)
FirehoseCommitMessage	레코드 작성·갱신·삭제 시
FirehoseIdentityMessage	identity 이벤트 시
FirehoseAccountMessage	계정 이벤트 시
FirehoseSyncMessage	저장소 동기화 이벤트 시

php artisan make:listener FirehosePostListener

namespace App\Listeners;

use Revolution\Bluesky\Events\Firehose\FirehoseCommitMessage;

class FirehosePostListener
{
    public function handle(FirehoseCommitMessage $event): void
    {
        // 컬렉션 종류 확인
        if ($event->collection !== 'app.bsky.feed.post') {
            return;
        }

        // 조작 종류: create / update / delete
        $action = $event->action;

        // 게시자의 DID
        $did = $event->did;

        // 레코드 내용(디코드된 배열)
        $record = $event->record;
        $text = data_get($record, 'value.text', '');

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

​

설정

config/bluesky.php에서 접속 호스트나 로그 설정을 변경할 수 있습니다.

// 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')),
    ],
],

.env에서의 설정 예시:

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

​

Labeler와의 조합

Labeler 서버와 Jetstream / Firehose를 동시에 기동할 수 있습니다. Labeler가 받은 라벨링 요청 처리에 Jetstream이나 Firehose의 데이터를 이용할 수 있습니다.

# Labeler + Jetstream(팔로우 이벤트를 감시)
php artisan bluesky:labeler:server start --jetstream -C app.bsky.graph.follow

# Labeler + Firehose(모든 데이터를 수신)
php artisan bluesky:labeler:server start --firehose

Labeler의 상세는 Labeler 페이지를 참조하세요.

​

장시간 실행 프로세스의 운용

WebSocket 명령어는 장시간 기동을 지속하는 프로세스입니다. 프로덕션 환경에서는 Supervisor 등의 프로세스 관리 툴을 사용하세요.

​

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

# Supervisor를 재로드하여 기동
sudo supervisorctl reread
sudo supervisorctl update
sudo supervisorctl start bluesky-jetstream:*

​

Laravel Forge의 데몬 설정

Laravel Forge를 사용하는 경우에는 Daemons 섹션에서 데몬을 추가합니다.

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

​

Laravel Cloud에서의 백그라운드 프로세스 설정

WebSocket 명령어는 WebSocket 클라이언트로서 Bluesky의 스트림에 접속하기 때문에, Laravel Cloud에서도 동작합니다. Laravel Cloud의 백그라운드 프로세스(커스텀 워커)로 설정하세요.

Laravel Cloud의 백그라운드 프로세스 설정에서 Custom Worker를 추가합니다.

Jetstream의 경우:

php artisan bluesky:ws start

Firehose의 경우:

php artisan bluesky:firehose start

배포 시 프로세스 정지·재기동 등은 모두 Laravel Cloud 측에서 자동으로 처리됩니다. 백그라운드 프로세스 설정 외에 추가 설정은 불필요합니다.

​

주의 사항

• 프로세스가 예기치 못하게 종료된 경우, autorestart=true로 자동 재기동됩니다.
• 메모리 누수를 방지하기 위해 정기적인 재기동을 검토해 주세요.
• 대량의 메시지를 수신하는 Firehose에서는 리스너 내부의 처리를 비동기(Queue Job)로 하는 것을 권장합니다.

// 리스너에서 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);
    }
}

Source: src/Console/WebSocket