Build Reactive Signals for Bluesky's AT Protocol Firehose in Laravel
Merge pull request #5 from socialdept/dev
Add GitHub Actions for automated testing and code style checks
authored by
Miguel Batres
and committed by
GitHub
8322e9d9
459c02a5
+164
-56
+26
.github/workflows/code-style.yml
+26
.github/workflows/code-style.yml
···
1
+
name: Code Style
2
+
3
+
on:
4
+
push:
5
+
branches: [ main, develop ]
6
+
pull_request:
7
+
branches: [ main, develop ]
8
+
9
+
jobs:
10
+
php-cs-fixer:
11
+
runs-on: ubuntu-latest
12
+
13
+
steps:
14
+
- name: Checkout code
15
+
uses: actions/checkout@v4
16
+
17
+
- name: Setup PHP
18
+
uses: shivammathur/setup-php@v2
19
+
with:
20
+
php-version: 8.3
21
+
extensions: gmp, mbstring, json
22
+
coverage: none
23
+
tools: php-cs-fixer
24
+
25
+
- name: Run PHP CS Fixer
26
+
run: php-cs-fixer fix --dry-run --diff --verbose
+32
.github/workflows/tests.yml
+32
.github/workflows/tests.yml
···
1
+
name: Tests
2
+
3
+
on:
4
+
push:
5
+
branches: [ main, develop ]
6
+
pull_request:
7
+
branches: [ main, develop ]
8
+
9
+
jobs:
10
+
test:
11
+
runs-on: ubuntu-latest
12
+
13
+
name: Tests (PHP 8.2 - Laravel 12)
14
+
15
+
steps:
16
+
- name: Checkout code
17
+
uses: actions/checkout@v4
18
+
19
+
- name: Setup PHP
20
+
uses: shivammathur/setup-php@v2
21
+
with:
22
+
php-version: 8.2
23
+
extensions: gmp, mbstring, json
24
+
coverage: none
25
+
26
+
- name: Install dependencies
27
+
run: |
28
+
composer require "laravel/framework:^12.0" "orchestra/testbench:^10.0" --no-interaction --no-update
29
+
composer update --prefer-stable --prefer-dist --no-interaction
30
+
31
+
- name: Execute tests
32
+
run: vendor/bin/phpunit
+35
.php-cs-fixer.php
+35
.php-cs-fixer.php
···
1
+
<?php
2
+
3
+
use PhpCsFixer\Config;
4
+
use PhpCsFixer\Finder;
5
+
6
+
$finder = Finder::create()
7
+
->in(__DIR__ . '/src')
8
+
->in(__DIR__ . '/tests')
9
+
->name('*.php')
10
+
->notName('*.blade.php')
11
+
->ignoreDotFiles(true)
12
+
->ignoreVCS(true);
13
+
14
+
return (new Config())
15
+
->setRules([
16
+
'@PSR12' => true,
17
+
'array_syntax' => ['syntax' => 'short'],
18
+
'ordered_imports' => ['sort_algorithm' => 'alpha'],
19
+
'no_unused_imports' => true,
20
+
'not_operator_with_successor_space' => true,
21
+
'trailing_comma_in_multiline' => true,
22
+
'phpdoc_scalar' => true,
23
+
'unary_operator_spaces' => true,
24
+
'binary_operator_spaces' => true,
25
+
'blank_line_before_statement' => [
26
+
'statements' => ['break', 'continue', 'declare', 'return', 'throw', 'try'],
27
+
],
28
+
'phpdoc_single_line_var_spacing' => true,
29
+
'phpdoc_var_without_name' => true,
30
+
'method_argument_space' => [
31
+
'on_multiline' => 'ensure_fully_multiline',
32
+
'keep_multiple_spaces_after_comma' => true,
33
+
],
34
+
])
35
+
->setFinder($finder);
+2
-1
composer.json
+2
-1
composer.json
+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
+16
-19
tests/Integration/FirehoseConsumerTest.php
+16
-19
tests/Integration/FirehoseConsumerTest.php
···
64
64
65
65
// 'time' key
66
66
$cbor .= "\x64time"; // Text string 'time'
67
-
$cbor .= "\x78\x182024-01-01T00:00:00Z"; // Text string (length 24)
67
+
$cbor .= "\x74" . "2024-01-01T00:00:00Z"; // Text string (length 20)
68
68
69
69
// 'ops' key
70
70
$cbor .= "\x63ops"; // Text string 'ops'
···
101
101
// Create a minimal CAR with header and one block
102
102
$car = '';
103
103
104
-
// CAR header (minimal)
105
-
$headerCbor = "\xA1\x67version\x01"; // {version: 1}
104
+
// CAR header (minimal) - {version: 1}
105
+
$headerCbor = "\xA1\x67version\x01";
106
106
$headerLength = strlen($headerCbor);
107
107
$car .= chr($headerLength) . $headerCbor;
108
108
109
-
// Create a block with CID
110
-
$blockData = "\xA1\x64test\x65value"; // {test: "value"}
109
+
// Create a block with CID and data
110
+
// Block data: {test: "value"}
111
+
$blockData = "\xA1\x64test\x65value";
112
+
113
+
// Create CID: version 1, codec 0x71 (dag-cbor), sha256 hash
111
114
$cid = CID::fromBinary("\x01\x71\x12\x20" . str_repeat("\x00", 32));
112
115
$cidBinary = $cid->toBinary();
113
-
$cidLength = strlen($cidBinary);
114
116
115
-
$block = chr($cidLength) . $cidBinary . $blockData;
116
-
$blockLength = strlen($block);
117
+
// In CAR format: varint(cid_length + data_length), CID bytes, data bytes
118
+
$totalLength = strlen($cidBinary) + strlen($blockData);
119
+
$car .= chr($totalLength) . $cidBinary . $blockData;
117
120
118
-
// Add varint-encoded block length
119
-
$car .= chr($blockLength) . $block;
121
+
// Parse blocks
122
+
$blocks = CAR::blockMap($car, 'did:plc:test');
120
123
121
-
// This should not throw an error
122
-
$blocks = [];
123
-
foreach (CAR::blockMap($car, 'did:plc:test') as $key => $value) {
124
-
$blocks[$key] = $value;
125
-
}
126
-
127
-
// Even if empty, it shouldn't crash
128
124
$this->assertIsArray($blocks);
125
+
$this->assertNotEmpty($blocks);
129
126
}
130
127
131
128
public function test_firehose_consumer_message_structure(): void
···
154
151
// Add required fields
155
152
$payload .= "\x66blocks\x40"; // 'blocks' => empty byte string
156
153
$payload .= "\x63ops\x80"; // 'ops' => []
157
-
$payload .= "\x64time\x78\x182024-01-01T00:00:00Z"; // 'time' => timestamp
154
+
$payload .= "\x64time\x74" . "2024-01-01T00:00:00Z"; // 'time' => timestamp
158
155
159
156
// Combine header + payload
160
157
$message = $header . $payload;
···
192
189
$payload .= "\x65since\x66origin"; // since: "origin"
193
190
$payload .= "\x66blocks\x40"; // blocks: b''
194
191
$payload .= "\x63ops\x80"; // ops: []
195
-
$payload .= "\x64time\x78\x182024-01-01T00:00:00Z"; // time: "2024-01-01T00:00:00Z"
192
+
$payload .= "\x64time\x74" . "2024-01-01T00:00:00Z"; // time: "2024-01-01T00:00:00Z"
196
193
197
194
$message = $header . $payload;
198
195
-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