Build Reactive Signals for Bluesky's AT Protocol Firehose in Laravel
Run entire package through `php-cs-fixer`
+52
-36
+3
-2
src/Binary/Reader.php
+3
-2
src/Binary/Reader.php
···
17
17
18
18
public function __construct(
19
19
private readonly string $data,
20
-
) {}
20
+
) {
21
+
}
21
22
22
23
/**
23
24
* Get current position in the data.
···
58
59
*/
59
60
public function peek(): int
60
61
{
61
-
if (!$this->hasMore()) {
62
+
if (! $this->hasMore()) {
62
63
throw new RuntimeException('Unexpected end of data');
63
64
}
64
65
+2
-3
src/CAR/BlockReader.php
+2
-3
src/CAR/BlockReader.php
···
7
7
use Generator;
8
8
use SocialDept\Signal\Binary\Reader;
9
9
use SocialDept\Signal\Core\CID;
10
-
use SocialDept\Signal\Core\CBOR;
11
10
12
11
/**
13
12
* CAR (Content Addressable aRchive) block reader.
···
49
48
*/
50
49
private function skipHeader(): void
51
50
{
52
-
if (!$this->reader->hasMore()) {
51
+
if (! $this->reader->hasMore()) {
53
52
return;
54
53
}
55
54
···
67
66
*/
68
67
private function readBlock(): ?array
69
68
{
70
-
if (!$this->reader->hasMore()) {
69
+
if (! $this->reader->hasMore()) {
71
70
return null;
72
71
}
73
72
+7
-6
src/CAR/RecordExtractor.php
+7
-6
src/CAR/RecordExtractor.php
···
5
5
namespace SocialDept\Signal\CAR;
6
6
7
7
use Generator;
8
-
use SocialDept\Signal\Core\CID;
9
8
use SocialDept\Signal\Core\CBOR;
9
+
use SocialDept\Signal\Core\CID;
10
10
11
11
/**
12
12
* Extract records from AT Protocol MST (Merkle Search Tree) blocks.
···
21
21
public function __construct(
22
22
private readonly array $blocks,
23
23
private readonly string $did,
24
-
) {}
24
+
) {
25
+
}
25
26
26
27
/**
27
28
* Extract all records from blocks.
···
47
48
$cidStr = $cid->toString();
48
49
49
50
// Get block data
50
-
if (!isset($this->blocks[$cidStr])) {
51
+
if (! isset($this->blocks[$cidStr])) {
51
52
// Block not found - might be a pruned tree, skip it
52
53
return;
53
54
}
···
57
58
// Decode CBOR block
58
59
$node = CBOR::decode($blockData);
59
60
60
-
if (!is_array($node)) {
61
+
if (! is_array($node)) {
61
62
return;
62
63
}
63
64
···
69
70
// Process entries
70
71
if (isset($node['e']) && is_array($node['e'])) {
71
72
foreach ($node['e'] as $entry) {
72
-
if (!is_array($entry)) {
73
+
if (! is_array($entry)) {
73
74
continue;
74
75
}
75
76
···
121
122
{
122
123
$cidStr = $cid->toString();
123
124
124
-
if (!isset($this->blocks[$cidStr])) {
125
+
if (! isset($this->blocks[$cidStr])) {
125
126
return null;
126
127
}
127
128
+3
-4
src/Core/CAR.php
+3
-4
src/Core/CAR.php
···
4
4
5
5
namespace SocialDept\Signal\Core;
6
6
7
-
use Generator;
8
7
use SocialDept\Signal\CAR\BlockReader;
9
-
use SocialDept\Signal\CAR\RecordExtractor;
10
8
11
9
/**
12
10
* CAR (Content Addressable aRchive) facade.
···
29
27
{
30
28
// Read all blocks from CAR
31
29
$blockReader = new BlockReader($data);
30
+
32
31
return $blockReader->getBlockMap();
33
32
}
34
33
···
46
45
47
46
$decoded = CBOR::decode($firstBlock);
48
47
49
-
if (!is_array($decoded)) {
48
+
if (! is_array($decoded)) {
50
49
return null;
51
50
}
52
51
···
67
66
68
67
$commit = CBOR::decode($firstBlock);
69
68
70
-
if (!is_array($commit)) {
69
+
if (! is_array($commit)) {
71
70
return null;
72
71
}
73
72
+2
-1
src/Core/CID.php
+2
-1
src/Core/CID.php
+2
-1
src/Events/AccountEvent.php
+2
-1
src/Events/AccountEvent.php
+2
-1
src/Events/IdentityEvent.php
+2
-1
src/Events/IdentityEvent.php
+2
-1
src/Events/SignalEvent.php
+2
-1
src/Events/SignalEvent.php
+6
-2
src/Jobs/ProcessSignalJob.php
+6
-2
src/Jobs/ProcessSignalJob.php
···
12
12
13
13
class ProcessSignalJob implements ShouldQueue
14
14
{
15
-
use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
15
+
use Dispatchable;
16
+
use InteractsWithQueue;
17
+
use Queueable;
18
+
use SerializesModels;
16
19
17
20
public function __construct(
18
21
protected Signal $signal,
19
22
protected SignalEvent $event,
20
-
) {}
23
+
) {
24
+
}
21
25
22
26
public function handle(): void
23
27
{
+2
-1
src/Services/FirehoseConsumer.php
+2
-1
src/Services/FirehoseConsumer.php
···
83
83
*/
84
84
protected function connect(string $url): void
85
85
{
86
-
$this->connection = new WebSocketConnection;
86
+
$this->connection = new WebSocketConnection();
87
87
88
88
// Set up event handlers
89
89
$this->connection
···
413
413
414
414
if ($this->reconnectAttempts >= $maxAttempts) {
415
415
Log::error('Signal: Max reconnection attempts reached');
416
+
416
417
throw new ConnectionException('Failed to reconnect to Firehose after '.$maxAttempts.' attempts');
417
418
}
418
419
+2
-1
src/Services/JetstreamConsumer.php
+2
-1
src/Services/JetstreamConsumer.php
···
76
76
*/
77
77
protected function connect(string $url): void
78
78
{
79
-
$this->connection = new WebSocketConnection;
79
+
$this->connection = new WebSocketConnection();
80
80
81
81
// Set up event handlers
82
82
$this->connection
···
178
178
179
179
if ($this->reconnectAttempts >= $maxAttempts) {
180
180
Log::error('Signal: Max reconnection attempts reached');
181
+
181
182
throw new ConnectionException('Failed to reconnect to Jetstream after '.$maxAttempts.' attempts');
182
183
}
183
184
+2
-1
src/Services/SignalManager.php
+2
-1
src/Services/SignalManager.php
+4
-4
src/SignalServiceProvider.php
+4
-4
src/SignalServiceProvider.php
···
27
27
// Register cursor store
28
28
$this->app->singleton(CursorStore::class, function ($app) {
29
29
return match (config('signal.cursor_storage')) {
30
-
'redis' => new RedisCursorStore,
31
-
'file' => new FileCursorStore,
32
-
default => new DatabaseCursorStore,
30
+
'redis' => new RedisCursorStore(),
31
+
'file' => new FileCursorStore(),
32
+
default => new DatabaseCursorStore(),
33
33
};
34
34
});
35
35
36
36
// Register signal registry
37
37
$this->app->singleton(SignalRegistry::class, function ($app) {
38
-
$registry = new SignalRegistry;
38
+
$registry = new SignalRegistry();
39
39
40
40
// Register configured signals
41
41
foreach (config('signal.signals', []) as $signal) {
+2
-2
src/Storage/FileCursorStore.php
+2
-2
src/Storage/FileCursorStore.php
···
15
15
16
16
// Ensure directory exists
17
17
$directory = dirname($this->path);
18
-
if (!File::exists($directory)) {
18
+
if (! File::exists($directory)) {
19
19
File::makeDirectory($directory, 0755, true);
20
20
}
21
21
}
22
22
23
23
public function get(): ?int
24
24
{
25
-
if (!File::exists($this->path)) {
25
+
if (! File::exists($this->path)) {
26
26
return null;
27
27
}
28
28
+7
-1
src/Support/WebSocketConnection.php
+7
-1
src/Support/WebSocketConnection.php
···
64
64
if ($this->onError) {
65
65
($this->onError)($e);
66
66
}
67
+
67
68
throw $e;
68
69
}
69
70
);
···
74
75
*/
75
76
public function send(string $message): bool
76
77
{
77
-
if (!$this->connected || !$this->connection) {
78
+
if (! $this->connected || ! $this->connection) {
78
79
return false;
79
80
}
80
81
81
82
try {
82
83
$this->connection->send($message);
84
+
83
85
return true;
84
86
} catch (\Exception $e) {
85
87
if ($this->onError) {
86
88
($this->onError)($e);
87
89
}
90
+
88
91
return false;
89
92
}
90
93
}
···
114
117
public function onMessage(callable $callback): self
115
118
{
116
119
$this->onMessage = $callback(...);
120
+
117
121
return $this;
118
122
}
119
123
···
123
127
public function onClose(callable $callback): self
124
128
{
125
129
$this->onClose = $callback(...);
130
+
126
131
return $this;
127
132
}
128
133
···
132
137
public function onError(callable $callback): self
133
138
{
134
139
$this->onError = $callback(...);
140
+
135
141
return $this;
136
142
}
137
143
-1
tests/Unit/SignalRegistryTest.php
-1
tests/Unit/SignalRegistryTest.php
+4
-4
tests/Unit/SignalTest.php
+4
-4
tests/Unit/SignalTest.php
···
12
12
/** @test */
13
13
public function it_can_create_a_signal()
14
14
{
15
-
$signal = new class extends Signal {
15
+
$signal = new class () extends Signal {
16
16
public function eventTypes(): array
17
17
{
18
18
return ['commit'];
···
31
31
/** @test */
32
32
public function it_can_filter_by_exact_collection()
33
33
{
34
-
$signal = new class extends Signal {
34
+
$signal = new class () extends Signal {
35
35
public function eventTypes(): array
36
36
{
37
37
return ['commit'];
···
66
66
/** @test */
67
67
public function it_can_filter_by_wildcard_collection()
68
68
{
69
-
$signalClass = new class extends Signal {
69
+
$signalClass = new class () extends Signal {
70
70
public function eventTypes(): array
71
71
{
72
72
return ['commit'];
···
84
84
};
85
85
86
86
// Create registry and register the signal
87
-
$registry = new \SocialDept\Signal\Services\SignalRegistry;
87
+
$registry = new \SocialDept\Signal\Services\SignalRegistry();
88
88
$registry->register($signalClass::class);
89
89
90
90
// Test that it matches app.bsky.feed.post