···11+<?php
22+33+declare(strict_types=1);
44+55+namespace SocialDept\Signal\Binary;
66+77+use RuntimeException;
88+99+/**
1010+ * Binary data reader with position tracking.
1111+ *
1212+ * Provides stream-like interface for reading from binary strings.
1313+ */
1414+class Reader
1515+{
1616+ private int $position = 0;
1717+1818+ public function __construct(
1919+ private readonly string $data,
2020+ ) {}
2121+2222+ /**
2323+ * Get current position in the data.
2424+ */
2525+ public function getPosition(): int
2626+ {
2727+ return $this->position;
2828+ }
2929+3030+ /**
3131+ * Get total length of data.
3232+ */
3333+ public function getLength(): int
3434+ {
3535+ return strlen($this->data);
3636+ }
3737+3838+ /**
3939+ * Check if there's more data to read.
4040+ */
4141+ public function hasMore(): bool
4242+ {
4343+ return $this->position < strlen($this->data);
4444+ }
4545+4646+ /**
4747+ * Get remaining bytes count.
4848+ */
4949+ public function remaining(): int
5050+ {
5151+ return strlen($this->data) - $this->position;
5252+ }
5353+5454+ /**
5555+ * Peek at the next byte without advancing position.
5656+ *
5757+ * @throws RuntimeException If no more data available
5858+ */
5959+ public function peek(): int
6060+ {
6161+ if (!$this->hasMore()) {
6262+ throw new RuntimeException('Unexpected end of data');
6363+ }
6464+6565+ return ord($this->data[$this->position]);
6666+ }
6767+6868+ /**
6969+ * Read a single byte and advance position.
7070+ *
7171+ * @throws RuntimeException If no more data available
7272+ */
7373+ public function readByte(): int
7474+ {
7575+ $byte = $this->peek();
7676+ $this->position++;
7777+7878+ return $byte;
7979+ }
8080+8181+ /**
8282+ * Read exactly N bytes and advance position.
8383+ *
8484+ * @throws RuntimeException If not enough data available
8585+ */
8686+ public function readBytes(int $length): string
8787+ {
8888+ if ($this->remaining() < $length) {
8989+ throw new RuntimeException("Cannot read {$length} bytes, only {$this->remaining()} remaining");
9090+ }
9191+9292+ $bytes = substr($this->data, $this->position, $length);
9393+ $this->position += $length;
9494+9595+ return $bytes;
9696+ }
9797+9898+ /**
9999+ * Read a varint (variable-length integer).
100100+ *
101101+ * @throws RuntimeException If varint is malformed
102102+ */
103103+ public function readVarint(): int
104104+ {
105105+ return Varint::decode($this->data, $this->position);
106106+ }
107107+108108+ /**
109109+ * Get all remaining data without advancing position.
110110+ */
111111+ public function peekRemaining(): string
112112+ {
113113+ return substr($this->data, $this->position);
114114+ }
115115+116116+ /**
117117+ * Read all remaining data and advance position to end.
118118+ */
119119+ public function readRemaining(): string
120120+ {
121121+ $remaining = $this->peekRemaining();
122122+ $this->position = strlen($this->data);
123123+124124+ return $remaining;
125125+ }
126126+127127+ /**
128128+ * Skip N bytes forward.
129129+ *
130130+ * @throws RuntimeException If trying to skip past end
131131+ */
132132+ public function skip(int $bytes): void
133133+ {
134134+ if ($this->remaining() < $bytes) {
135135+ throw new RuntimeException("Cannot skip {$bytes} bytes, only {$this->remaining()} remaining");
136136+ }
137137+138138+ $this->position += $bytes;
139139+ }
140140+}
+68
src/Binary/Varint.php
···11+<?php
22+33+declare(strict_types=1);
44+55+namespace SocialDept\Signal\Binary;
66+77+use RuntimeException;
88+99+/**
1010+ * Varint (Variable-Length Integer) decoder for unsigned integers.
1111+ *
1212+ * Used in CAR format to encode block lengths. Implements the same
1313+ * varint encoding used in Protocol Buffers and other binary formats.
1414+ */
1515+class Varint
1616+{
1717+ /**
1818+ * Decode a varint from the beginning of the data.
1919+ *
2020+ * @param string $data Binary data containing varint
2121+ * @param int $offset Starting position (will be updated to position after varint)
2222+ * @return int Decoded unsigned integer
2323+ * @throws RuntimeException If varint is malformed or data ends unexpectedly
2424+ */
2525+ public static function decode(string $data, int &$offset = 0): int
2626+ {
2727+ $result = 0;
2828+ $shift = 0;
2929+ $length = strlen($data);
3030+3131+ while ($offset < $length) {
3232+ $byte = ord($data[$offset]);
3333+ $offset++;
3434+3535+ // Take lower 7 bits and shift into result
3636+ $result |= ($byte & 0x7F) << $shift;
3737+3838+ // If MSB is 0, we're done
3939+ if (($byte & 0x80) === 0) {
4040+ return $result;
4141+ }
4242+4343+ $shift += 7;
4444+4545+ // Prevent overflow (64-bit max)
4646+ if ($shift > 63) {
4747+ throw new RuntimeException('Varint too long (max 64 bits)');
4848+ }
4949+ }
5050+5151+ throw new RuntimeException('Unexpected end of varint data');
5252+ }
5353+5454+ /**
5555+ * Read a varint and return both the value and remaining data.
5656+ *
5757+ * @param string $data Binary data starting with varint
5858+ * @return array{0: int, 1: string} [decoded value, remaining data]
5959+ */
6060+ public static function decodeFirst(string $data): array
6161+ {
6262+ $offset = 0;
6363+ $value = self::decode($data, $offset);
6464+ $remainder = substr($data, $offset);
6565+6666+ return [$value, $remainder];
6767+ }
6868+}
+133
src/CAR/BlockReader.php
···11+<?php
22+33+declare(strict_types=1);
44+55+namespace SocialDept\Signal\CAR;
66+77+use Generator;
88+use SocialDept\Signal\Binary\Reader;
99+use SocialDept\Signal\Core\CID;
1010+use SocialDept\Signal\Core\CBOR;
1111+1212+/**
1313+ * CAR (Content Addressable aRchive) block reader.
1414+ *
1515+ * Reads blocks from CAR format data used in AT Protocol commits.
1616+ */
1717+class BlockReader
1818+{
1919+ private Reader $reader;
2020+2121+ public function __construct(string $data)
2222+ {
2323+ $this->reader = new Reader($data);
2424+ }
2525+2626+ /**
2727+ * Read all blocks from CAR data.
2828+ *
2929+ * Yields [CID, block data] pairs.
3030+ *
3131+ * @return Generator<array{0: CID, 1: string}>
3232+ */
3333+ public function blocks(): Generator
3434+ {
3535+ // Skip CAR header (we don't need it for Firehose processing)
3636+ $this->skipHeader();
3737+3838+ // Read blocks until end of data
3939+ while ($this->reader->hasMore()) {
4040+ $block = $this->readBlock();
4141+ if ($block !== null) {
4242+ yield $block;
4343+ }
4444+ }
4545+ }
4646+4747+ /**
4848+ * Skip CAR header.
4949+ */
5050+ private function skipHeader(): void
5151+ {
5252+ if (!$this->reader->hasMore()) {
5353+ return;
5454+ }
5555+5656+ // Read header length (varint)
5757+ $headerLength = $this->reader->readVarint();
5858+5959+ // Skip header data
6060+ $this->reader->skip($headerLength);
6161+ }
6262+6363+ /**
6464+ * Read a single block.
6565+ *
6666+ * @return array{0: CID, 1: string}|null [CID, block data] or null if no more blocks
6767+ */
6868+ private function readBlock(): ?array
6969+ {
7070+ if (!$this->reader->hasMore()) {
7171+ return null;
7272+ }
7373+7474+ // Read block length (varint) - this is the total length of CID + data
7575+ $blockLength = $this->reader->readVarint();
7676+7777+ if ($blockLength === 0) {
7878+ return null;
7979+ }
8080+8181+ // Read entire block data
8282+ $blockData = $this->reader->readBytes($blockLength);
8383+8484+ // Parse CID from the beginning of block data
8585+ // CIDs in CAR blocks are self-delimiting (no separate length prefix)
8686+ // We need to parse the CID to find out its length
8787+ $cidReader = new Reader($blockData);
8888+8989+ // Read CID version
9090+ $version = $cidReader->readVarint();
9191+9292+ if ($version === 0x12) {
9393+ // CIDv0 - multihash only (starting with 0x12 for SHA-256)
9494+ $hashLength = $cidReader->readVarint();
9595+ $cidReader->readBytes($hashLength); // Skip hash bytes
9696+ } elseif ($version === 1) {
9797+ // CIDv1 - version + codec + multihash
9898+ $codec = $cidReader->readVarint();
9999+ $hashType = $cidReader->readVarint();
100100+ $hashLength = $cidReader->readVarint();
101101+ $cidReader->readBytes($hashLength); // Skip hash bytes
102102+ } else {
103103+ throw new \RuntimeException("Unsupported CID version in CAR block: {$version}");
104104+ }
105105+106106+ // Now we know the CID length
107107+ $cidLength = $cidReader->getPosition();
108108+ $cidBytes = substr($blockData, 0, $cidLength);
109109+ $cid = CID::fromBinary($cidBytes);
110110+111111+ // Remaining data is the block content
112112+ $content = substr($blockData, $cidLength);
113113+114114+ return [$cid, $content];
115115+ }
116116+117117+ /**
118118+ * Get all blocks as an associative array.
119119+ *
120120+ * @return array<string, string> Map of CID string => block data
121121+ */
122122+ public function getBlockMap(): array
123123+ {
124124+ $blocks = [];
125125+126126+ foreach ($this->blocks() as [$cid, $data]) {
127127+ $cidString = $cid->toString();
128128+ $blocks[$cidString] = $data;
129129+ }
130130+131131+ return $blocks;
132132+ }
133133+}
+132
src/CAR/RecordExtractor.php
···11+<?php
22+33+declare(strict_types=1);
44+55+namespace SocialDept\Signal\CAR;
66+77+use Generator;
88+use SocialDept\Signal\Core\CID;
99+use SocialDept\Signal\Core\CBOR;
1010+1111+/**
1212+ * Extract records from AT Protocol MST (Merkle Search Tree) blocks.
1313+ *
1414+ * Walks MST structure to extract collection/rkey records with their values.
1515+ */
1616+class RecordExtractor
1717+{
1818+ /**
1919+ * @param array<string, string> $blocks Map of CID string => block data
2020+ */
2121+ public function __construct(
2222+ private readonly array $blocks,
2323+ private readonly string $did,
2424+ ) {}
2525+2626+ /**
2727+ * Extract all records from blocks.
2828+ *
2929+ * Yields records in format: "collection/rkey" => record data
3030+ *
3131+ * @return Generator<string, array>
3232+ */
3333+ public function extractRecords(CID $rootCid): Generator
3434+ {
3535+ yield from $this->walkTree($rootCid, '');
3636+ }
3737+3838+ /**
3939+ * Recursively walk MST tree.
4040+ *
4141+ * @param CID $cid Current node CID
4242+ * @param string $prefix Path prefix accumulated from parent nodes
4343+ * @return Generator<string, array>
4444+ */
4545+ private function walkTree(CID $cid, string $prefix): Generator
4646+ {
4747+ $cidStr = $cid->toString();
4848+4949+ // Get block data
5050+ if (!isset($this->blocks[$cidStr])) {
5151+ // Block not found - might be a pruned tree, skip it
5252+ return;
5353+ }
5454+5555+ $blockData = $this->blocks[$cidStr];
5656+5757+ // Decode CBOR block
5858+ $node = CBOR::decode($blockData);
5959+6060+ if (!is_array($node)) {
6161+ return;
6262+ }
6363+6464+ // Process left subtree if exists
6565+ if (isset($node['l']) && $node['l'] instanceof CID) {
6666+ yield from $this->walkTree($node['l'], $prefix);
6767+ }
6868+6969+ // Process entries
7070+ if (isset($node['e']) && is_array($node['e'])) {
7171+ foreach ($node['e'] as $entry) {
7272+ if (!is_array($entry)) {
7373+ continue;
7474+ }
7575+7676+ // Build full key from prefix + entry key
7777+ $entryPrefix = $entry['p'] ?? 0;
7878+ $keyPart = $entry['k'] ?? '';
7979+ $fullKey = substr($prefix, 0, $entryPrefix) . $keyPart;
8080+8181+ // If entry has a tree link, walk it
8282+ if (isset($entry['t']) && $entry['t'] instanceof CID) {
8383+ yield from $this->walkTree($entry['t'], $fullKey);
8484+ }
8585+8686+ // If entry has a value (record), yield it
8787+ if (isset($entry['v']) && $entry['v'] instanceof CID) {
8888+ $recordCid = $entry['v'];
8989+ $record = $this->getRecord($recordCid);
9090+9191+ if ($record !== null) {
9292+ // Parse collection/rkey from key
9393+ $parts = explode('/', $fullKey, 2);
9494+ if (count($parts) === 2) {
9595+ [$collection, $rkey] = $parts;
9696+ $path = "{$collection}/{$rkey}";
9797+9898+ yield $path => [
9999+ 'uri' => "at://{$this->did}/{$path}",
100100+ 'cid' => $recordCid->toString(),
101101+ 'value' => $record,
102102+ ];
103103+ } else {
104104+ // Debug: log when key format doesn't match expected pattern
105105+ \Illuminate\Support\Facades\Log::debug('Signal: MST key parse failed', [
106106+ 'fullKey' => $fullKey,
107107+ 'parts' => $parts,
108108+ 'did' => $this->did,
109109+ ]);
110110+ }
111111+ }
112112+ }
113113+ }
114114+ }
115115+ }
116116+117117+ /**
118118+ * Get record data from block.
119119+ */
120120+ private function getRecord(CID $cid): ?array
121121+ {
122122+ $cidStr = $cid->toString();
123123+124124+ if (!isset($this->blocks[$cidStr])) {
125125+ return null;
126126+ }
127127+128128+ $data = CBOR::decode($this->blocks[$cidStr]);
129129+130130+ return is_array($data) ? $data : null;
131131+ }
132132+}
···6666 /** @test */
6767 public function it_can_filter_by_wildcard_collection()
6868 {
6969- $signal = new class extends Signal {
6969+ $signalClass = new class extends Signal {
7070 public function eventTypes(): array
7171 {
7272 return ['commit'];
···8383 }
8484 };
85858686+ // Create registry and register the signal
8787+ $registry = new \SocialDept\Signal\Services\SignalRegistry;
8888+ $registry->register($signalClass::class);
8989+8690 // Test that it matches app.bsky.feed.post
8791 $postEvent = new SignalEvent(
8892 did: 'did:plc:test',
···96100 ),
97101 );
981029999- $this->assertTrue($signal->shouldHandle($postEvent));
103103+ $matchingSignals = $registry->getMatchingSignals($postEvent);
104104+ $this->assertCount(1, $matchingSignals);
100105101106 // Test that it matches app.bsky.feed.like
102107 $likeEvent = new SignalEvent(
···111116 ),
112117 );
113118114114- $this->assertTrue($signal->shouldHandle($likeEvent));
119119+ $matchingSignals = $registry->getMatchingSignals($likeEvent);
120120+ $this->assertCount(1, $matchingSignals);
115121116122 // Test that it does NOT match app.bsky.graph.follow
117123 $followEvent = new SignalEvent(
···126132 ),
127133 );
128134129129- $this->assertFalse($signal->shouldHandle($followEvent));
135135+ $matchingSignals = $registry->getMatchingSignals($followEvent);
136136+ $this->assertCount(0, $matchingSignals);
130137 }
131138}
+75
tests/Unit/VarintTest.php
···11+<?php
22+33+declare(strict_types=1);
44+55+namespace SocialDept\Signal\Tests\Unit;
66+77+use PHPUnit\Framework\TestCase;
88+use RuntimeException;
99+use SocialDept\Signal\Binary\Varint;
1010+1111+class VarintTest extends TestCase
1212+{
1313+ public function test_decode_single_byte_values(): void
1414+ {
1515+ $this->assertSame(0, Varint::decode("\x00"));
1616+ $this->assertSame(1, Varint::decode("\x01"));
1717+ $this->assertSame(127, Varint::decode("\x7F"));
1818+ }
1919+2020+ public function test_decode_multi_byte_values(): void
2121+ {
2222+ // 128 = 0x80 0x01
2323+ $this->assertSame(128, Varint::decode("\x80\x01"));
2424+2525+ // 300 = 0xAC 0x02
2626+ $this->assertSame(300, Varint::decode("\xAC\x02"));
2727+2828+ // 16384 = 0x80 0x80 0x01
2929+ $this->assertSame(16384, Varint::decode("\x80\x80\x01"));
3030+ }
3131+3232+ public function test_decode_with_offset(): void
3333+ {
3434+ $data = "\x00\x01\x7F\x80\x01";
3535+ $offset = 0;
3636+3737+ $this->assertSame(0, Varint::decode($data, $offset));
3838+ $this->assertSame(1, $offset);
3939+4040+ $this->assertSame(1, Varint::decode($data, $offset));
4141+ $this->assertSame(2, $offset);
4242+4343+ $this->assertSame(127, Varint::decode($data, $offset));
4444+ $this->assertSame(3, $offset);
4545+4646+ $this->assertSame(128, Varint::decode($data, $offset));
4747+ $this->assertSame(5, $offset);
4848+ }
4949+5050+ public function test_decode_first_returns_value_and_remainder(): void
5151+ {
5252+ [$value, $remainder] = Varint::decodeFirst("\x7F\x01\x02\x03");
5353+5454+ $this->assertSame(127, $value);
5555+ $this->assertSame("\x01\x02\x03", $remainder);
5656+ }
5757+5858+ public function test_decode_throws_on_unexpected_end(): void
5959+ {
6060+ $this->expectException(RuntimeException::class);
6161+ $this->expectExceptionMessage('Unexpected end of varint data');
6262+6363+ Varint::decode("\x80");
6464+ }
6565+6666+ public function test_decode_throws_on_too_long_varint(): void
6767+ {
6868+ $this->expectException(RuntimeException::class);
6969+ $this->expectExceptionMessage('Varint too long (max 64 bits)');
7070+7171+ // Create a varint that would be longer than 64 bits (10 bytes with continuation bits)
7272+ $tooLong = str_repeat("\xFF", 10);
7373+ Varint::decode($tooLong);
7474+ }
7575+}