a collection of lightweight TypeScript packages for AT Protocol, the protocol powering Bluesky
atproto
bluesky
typescript
npm
feat(car): v4, and streaming repo
+866
-439
+39
.changeset/few-kings-lose.md
+39
.changeset/few-kings-lose.md
···
1
+
---
2
+
'@atcute/car': minor
3
+
---
4
+
5
+
add streaming support for `CarReader` and `RepoReader`, which should allow for efficient reading of
6
+
CAR archives.
7
+
8
+
```ts
9
+
import { CarReader, RepoReader } from '@atcute/car/v4';
10
+
11
+
// read AT Protocol repository exports
12
+
{
13
+
await using repo = RepoReader.fromStream(stream);
14
+
15
+
for await (const entry of repo) {
16
+
entry;
17
+
// ^? RepoEntry { collection: 'app.bsky.feed.post', rkey: '3lprcc55bb222', ... }
18
+
}
19
+
20
+
repo.missingBlocks;
21
+
// ^? []
22
+
}
23
+
24
+
// read generic CAR archives
25
+
{
26
+
await using car = CarReader.fromStream(stream);
27
+
28
+
const roots = await car.roots();
29
+
30
+
for await (const entry of car) {
31
+
entry;
32
+
// ^? CarEntry { cid: CidLink {}, bytes: Uint8Array {}, ... }
33
+
}
34
+
}
35
+
```
36
+
37
+
please note that the reference PDS implementation does not yet support the Sync v1.1 proposal, which
38
+
would enable more efficient streaming. additionally, some PDSes may send valid but heavily
39
+
out-of-order archives that could impact streaming performance.
+58
-1
.changeset/silver-waves-grow.md
+58
-1
.changeset/silver-waves-grow.md
···
2
2
'@atcute/car': minor
3
3
---
4
4
5
-
streamed CAR reader
5
+
reorganized the exported functions, the new exports should be inline with other utility packages
6
+
from atcute.
7
+
8
+
normally this would be considered a breaking change, but because the change doesn't exactly change
9
+
any of the API, I've decided to turn this into a minor change.
10
+
11
+
to migrate, you'd need to change your imports from `@atcute/car` to `@atcute/car/v4` subpath.
12
+
13
+
```ts
14
+
// before (v3)
15
+
import { readCar, iterateAtpRepo } from '@atcute/car';
16
+
17
+
// read AT Protocol repository exports
18
+
for (const entry of iterateAtpRepo(buffer)) {
19
+
entry;
20
+
// ^? RepoEntry { ... }
21
+
}
22
+
23
+
// read generic CAR archives
24
+
{
25
+
const car = readCar(buffer);
26
+
const header = car.header;
27
+
28
+
for (const entry of car.iterate()) {
29
+
entry;
30
+
// ^? CarEntry { ... }
31
+
}
32
+
}
33
+
```
34
+
35
+
```ts
36
+
// after (v4)
37
+
import { CarReader, RepoReader } from '@atcute/car/v4';
38
+
39
+
// alternatively
40
+
import * as CarReader from '@atcute/car/v4/car-reader';
41
+
import * as RepoReader from '@atcute/car/v4/repo-reader';
42
+
43
+
// read AT Protocol repository exports
44
+
{
45
+
for (const entry of RepoReader.fromUint8Array(buffer)) {
46
+
entry;
47
+
// ^? RepoEntry { ... }
48
+
}
49
+
}
50
+
51
+
// read generic CAR archives
52
+
{
53
+
const car = CarReader.fromUint8Array(buffer);
54
+
const header = car.header;
55
+
56
+
// use for..of on `car` directly, `.iterate()` is deprecated
57
+
for (const entry of car) {
58
+
entry;
59
+
// ^? CarEntry { ... }
60
+
}
61
+
}
62
+
```
+22
-7
packages/utilities/car/README.md
+22
-7
packages/utilities/car/README.md
···
6
6
[dasl-car]: https://dasl.ing/car.html
7
7
8
8
```ts
9
-
// read through a CAR archive
10
-
const { header, iterate } = readCar(buf);
9
+
import { CarReader, RepoReader } from '@atcute/car/v4';
10
+
11
+
// read AT Protocol repository exports
12
+
{
13
+
await using repo = RepoReader.fromStream(stream);
14
+
15
+
for await (const entry of repo) {
16
+
entry;
17
+
// ^? RepoEntry { collection: 'app.bsky.feed.post', rkey: '3lprcc55bb222', ... }
18
+
}
11
19
12
-
for (const { cid, bytes } of iterate()) {
13
-
// ...
20
+
repo.missingBlocks;
21
+
// ^? []
14
22
}
15
23
16
-
// convenient iterator for reading through an AT Protocol CAR repository
17
-
for (const { collection, rkey, record } of iterateAtpRepo(buf)) {
18
-
// ...
24
+
// read generic CAR archives
25
+
{
26
+
await using car = CarReader.fromStream(stream);
27
+
28
+
const roots = await car.roots();
29
+
30
+
for await (const entry of car) {
31
+
entry;
32
+
// ^? CarEntry { cid: CidLink {}, bytes: Uint8Array {}, ... }
33
+
}
19
34
}
20
35
```
-77
packages/utilities/car/lib/atproto-repo.test.ts
-77
packages/utilities/car/lib/atproto-repo.test.ts
···
1
-
import { expect, it } from 'bun:test';
2
-
3
-
import { fromCidLink, toString } from '@atcute/cid';
4
-
import { fromBase64 } from '@atcute/multibase';
5
-
6
-
import { iterateAtpRepo } from './atproto-repo.js';
7
-
8
-
it('decodes atproto car files', () => {
9
-
const buf = fromBase64(
10
-
'OqJlcm9vdHOB2CpYJQABcRIgkD8I0DL+GsJ3OKREpf9k73yHguuSEYzEiXPGueoJg8FndmVy' +
11
-
'c2lvbgGPAQFxEiDqG8o/D37K3hldhQTMRq9/Uvyf7X9evn9eB9ZdgpYq6qRlJHR5cGV2YXBw' +
12
-
'LmJza3kuYWN0b3IucHJvZmlsZWljcmVhdGVkQXR4GDIwMjQtMDItMjRUMTI6MTU6NDEuMjE5' +
13
-
'WmtkZXNjcmlwdGlvbm90ZXN0aW5nIGFjY291bnRrZGlzcGxheU5hbWVg4AEBcRIgkD8I0DL+' +
14
-
'GsJ3OKREpf9k73yHguuSEYzEiXPGueoJg8GmY2RpZHggZGlkOnBsYzpzcmNxb3UybTd1cXVv' +
15
-
'Z3lkNXhrNGI1eTVjcmV2bTNsNXE1ZmplbnRjMmRjc2lnWEDeWWEO5/vV6SmnbUrLRu9WhWqI' +
16
-
'kHKANGFOin3xqFc4fgtuYzkbFXFJDMQU06nBWxict8FQ8Kas9Mr2fDAh++vVZGRhdGHYKlgl' +
17
-
'AAFxEiB2ibkpj3r4cdTag9v2ipIe8fxyjUFOgCjZbtYnfhyJ2GRwcmV29md2ZXJzaW9uA6QB' +
18
-
'AXESIHaJuSmPevhx1NqD2/aKkh7x/HKNQU6AKNlu1id+HInYomFlgaRha1gbYXBwLmJza3ku' +
19
-
'YWN0b3IucHJvZmlsZS9zZWxmYXAAYXTYKlglAAFxEiBvSJJSaF/w/fee+UmoLV84FDwZRC7p' +
20
-
'pJX484MghY0rM2F22CpYJQABcRIg6hvKPw9+yt4ZXYUEzEavf1L8n+1/Xr5/XgfWXYKWKuph' +
21
-
'bPaBAQFxEiBvSJJSaF/w/fee+UmoLV84FDwZRC7ppJX484MghY0rM6JhZYGkYWtYIGFwcC5i' +
22
-
'c2t5LmZlZWQucG9zdC8za201eW1rNGhoazJ6YXAAYXT2YXbYKlglAAFxEiDj+gU903L3F3Ar' +
23
-
'WCg+aeQZYEiM3ooIxqHbVvbQPZvEbGFs9qECAXESIOP6BT3TcvcXcCtYKD5p5BlgSIzeigjG' +
24
-
'odtW9tA9m8RspWR0ZXh0dWJlZXAgYm9vcCBAbWFyeS5teS5pZGUkdHlwZXJhcHAuYnNreS5m' +
25
-
'ZWVkLnBvc3RlbGFuZ3OBYmVuZmZhY2V0c4GjZSR0eXBld2FwcC5ic2t5LnJpY2h0ZXh0LmZh' +
26
-
'Y2V0ZWluZGV4omdieXRlRW5kFWlieXRlU3RhcnQKaGZlYXR1cmVzgaJjZGlkeCBkaWQ6cGxj' +
27
-
'OmlhNzZrdm5uZGp1dGdlZGdneDJpYnJlbWUkdHlwZXgfYXBwLmJza3kucmljaHRleHQuZmFj' +
28
-
'ZXQjbWVudGlvbmljcmVhdGVkQXR4GDIwMjQtMDItMjRUMTI6MTY6MjAuNjM3Wg',
29
-
);
30
-
31
-
const result = Array.from(iterateAtpRepo(buf), (entry) => ({
32
-
collection: entry.collection,
33
-
rkey: entry.rkey,
34
-
cid: toString(fromCidLink(entry.cid)),
35
-
record: entry.record,
36
-
}));
37
-
38
-
expect(result).toEqual([
39
-
{
40
-
collection: 'app.bsky.actor.profile',
41
-
rkey: 'self',
42
-
cid: 'bafyreihkdpfd6d36zlpbsxmfatgenl37kl6j73l7l27h6xqh2zoyffrk5i',
43
-
record: {
44
-
$type: 'app.bsky.actor.profile',
45
-
createdAt: '2024-02-24T12:15:41.219Z',
46
-
displayName: '',
47
-
description: 'testing account',
48
-
},
49
-
},
50
-
{
51
-
collection: 'app.bsky.feed.post',
52
-
rkey: '3km5ymk4hhk2z',
53
-
cid: 'bafyreihd7ict3u3s64lxak2yfa7gtzazmbeizxukbddkdw2w63id3g6enq',
54
-
record: {
55
-
$type: 'app.bsky.feed.post',
56
-
createdAt: '2024-02-24T12:16:20.637Z',
57
-
langs: ['en'],
58
-
text: 'beep boop @mary.my.id',
59
-
facets: [
60
-
{
61
-
$type: 'app.bsky.richtext.facet',
62
-
index: {
63
-
byteEnd: 21,
64
-
byteStart: 10,
65
-
},
66
-
features: [
67
-
{
68
-
did: 'did:plc:ia76kvnndjutgedggx2ibrem',
69
-
$type: 'app.bsky.richtext.facet#mention',
70
-
},
71
-
],
72
-
},
73
-
],
74
-
},
75
-
},
76
-
]);
77
-
});
-248
packages/utilities/car/lib/atproto-repo.ts
-248
packages/utilities/car/lib/atproto-repo.ts
···
1
-
import * as CBOR from '@atcute/cbor';
2
-
import * as CID from '@atcute/cid';
3
-
4
-
import { readCar, type CarEntry } from './reader.js';
5
-
6
-
const decoder = new TextDecoder();
7
-
8
-
export type BlockMap = Map<string, CarEntry>;
9
-
10
-
export class RepoEntry {
11
-
constructor(
12
-
/** The collection this record belongs to */
13
-
public readonly collection: string,
14
-
/** Record key */
15
-
public readonly rkey: string,
16
-
/** CID of this record */
17
-
public readonly cid: CID.CidLink,
18
-
private blockmap: BlockMap,
19
-
) {}
20
-
21
-
/**
22
-
* returns the associated CarEntry for this record
23
-
*/
24
-
get carEntry(): CarEntry {
25
-
const cid = this.cid.$link;
26
-
27
-
const entry = this.blockmap.get(cid);
28
-
assert(entry != null, `cid not found in blockmap; cid=${cid}`);
29
-
30
-
return entry;
31
-
}
32
-
33
-
/**
34
-
* returns the raw contents of this record
35
-
*/
36
-
get bytes(): Uint8Array {
37
-
return this.carEntry.bytes;
38
-
}
39
-
40
-
/**
41
-
* returns the decoded contents of this record
42
-
*/
43
-
get record(): unknown {
44
-
return CBOR.decode(this.bytes);
45
-
}
46
-
}
47
-
48
-
export function* iterateAtpRepo(buf: Uint8Array): Generator<RepoEntry> {
49
-
const { header, iterate } = readCar(buf);
50
-
const roots = header.data.roots;
51
-
52
-
assert(roots.length === 1, `expected only 1 root in the car archive; got=${roots.length}`);
53
-
54
-
const blockmap = collectBlock(iterate());
55
-
assert(blockmap.size > 0, `expected at least 1 block in the archive; got=${blockmap.size}`);
56
-
57
-
const commit = readBlock(blockmap, roots[0], isCommit);
58
-
for (const { key, cid } of walkMstEntries(blockmap, commit.data)) {
59
-
const [collection, rkey] = key.split('/');
60
-
61
-
yield new RepoEntry(collection, rkey, cid, blockmap);
62
-
}
63
-
}
64
-
65
-
/**
66
-
* collects entries from a CAR archive into a mapping of CID string -> actual bytes
67
-
* @param iterator a generator that yields objects with a `cid` and `bytes` property
68
-
* @returns a mapping of CID string -> actual bytes
69
-
*/
70
-
export function collectBlock(iterator: Generator<CarEntry>): BlockMap {
71
-
const blockmap: BlockMap = new Map();
72
-
for (const entry of iterator) {
73
-
blockmap.set(CID.toString(entry.cid), entry);
74
-
}
75
-
76
-
return blockmap;
77
-
}
78
-
79
-
/**
80
-
* reads a block from the blockmap and validates it against the provided validation function
81
-
* @param map a mapping of CID string -> actual bytes
82
-
* @param link a CID link to read
83
-
* @param validate a validation function to validate the decoded data
84
-
* @returns the decoded and validated data
85
-
*/
86
-
export function readBlock<T>(map: BlockMap, link: CID.CidLink, validate: (value: unknown) => value is T): T {
87
-
const cid = link.$link;
88
-
89
-
const entry = map.get(cid);
90
-
assert(entry != null, `cid not found in blockmap; cid=${cid}`);
91
-
92
-
const data = CBOR.decode(entry.bytes);
93
-
assert(validate(data), `validation failed for cid=${cid}`);
94
-
95
-
return data;
96
-
}
97
-
98
-
/** node entry object */
99
-
export interface NodeEntry {
100
-
key: string;
101
-
cid: CID.CidLink;
102
-
}
103
-
104
-
/**
105
-
* walks the entries of a Merkle Sorted Tree (MST) in a depth-first manner
106
-
* @param map a mapping of CID string -> actual bytes
107
-
* @param pointer a CID link to the root of the MST
108
-
* @returns a generator that yields the entries of the MST
109
-
*/
110
-
export function* walkMstEntries(map: BlockMap, pointer: CID.CidLink): Generator<NodeEntry> {
111
-
const data = readBlock(map, pointer, isMstNode);
112
-
const entries = data.e;
113
-
114
-
let lastKey = '';
115
-
116
-
if (data.l !== null) {
117
-
yield* walkMstEntries(map, data.l);
118
-
}
119
-
120
-
for (let i = 0, il = entries.length; i < il; i++) {
121
-
const entry = entries[i];
122
-
123
-
const key_str = decoder.decode(CBOR.fromBytes(entry.k));
124
-
const key = lastKey.slice(0, entry.p) + key_str;
125
-
126
-
lastKey = key;
127
-
128
-
yield { key: key, cid: entry.v };
129
-
130
-
if (entry.t !== null) {
131
-
yield* walkMstEntries(map, entry.t);
132
-
}
133
-
}
134
-
}
135
-
136
-
function assert(condition: boolean, message: string): asserts condition {
137
-
if (!condition) {
138
-
throw new Error(message);
139
-
}
140
-
}
141
-
142
-
const isCidLink = (value: unknown): value is CID.CidLink => {
143
-
if (value instanceof CID.CidLinkWrapper) {
144
-
return true;
145
-
}
146
-
147
-
if (value === null || typeof value !== 'object') {
148
-
return false;
149
-
}
150
-
151
-
return '$link' in value && typeof value.$link === 'string';
152
-
};
153
-
154
-
const isBytes = (value: unknown): value is CBOR.Bytes => {
155
-
if (value instanceof CBOR.BytesWrapper) {
156
-
return true;
157
-
}
158
-
159
-
if (value === null || typeof value !== 'object') {
160
-
return false;
161
-
}
162
-
163
-
return '$bytes' in value && typeof value.$bytes === 'string';
164
-
};
165
-
166
-
/** commit object */
167
-
export interface Commit {
168
-
version: 3;
169
-
did: string;
170
-
data: CID.CidLink;
171
-
rev: string;
172
-
prev: CID.CidLink | null;
173
-
sig: CBOR.Bytes;
174
-
}
175
-
176
-
/**
177
-
* checks if a value is a valid commit object
178
-
* @param value the value to check
179
-
* @returns true if the value is a valid commit object, false otherwise
180
-
*/
181
-
export const isCommit = (value: unknown): value is Commit => {
182
-
if (value === null || typeof value !== 'object') {
183
-
return false;
184
-
}
185
-
186
-
const obj = value as Record<string, unknown>;
187
-
188
-
return (
189
-
obj.version === 3 &&
190
-
typeof obj.did === 'string' &&
191
-
isCidLink(obj.data) &&
192
-
typeof obj.rev === 'string' &&
193
-
(obj.prev === null || isCidLink(obj.prev)) &&
194
-
isBytes(obj.sig)
195
-
);
196
-
};
197
-
198
-
/** mst tree entry object */
199
-
export interface TreeEntry {
200
-
/** count of bytes shared with previous TreeEntry in this Node (if any) */
201
-
p: number;
202
-
/** remainder of key for this TreeEntry, after "prefixlen" have been removed */
203
-
k: CBOR.Bytes;
204
-
/** link to a sub-tree Node at a lower level which has keys sorting after this TreeEntry's key (to the "right"), but before the next TreeEntry's key in this Node (if any) */
205
-
v: CID.CidLink;
206
-
/** next subtree (to the right of leaf) */
207
-
t: CID.CidLink | null;
208
-
}
209
-
210
-
/**
211
-
* checks if a value is a valid mst tree entry object
212
-
* @param value the value to check
213
-
* @returns true if the value is a valid mst tree entry object, false otherwise
214
-
*/
215
-
export const isTreeEntry = (value: unknown): value is TreeEntry => {
216
-
if (value === null || typeof value !== 'object') {
217
-
return false;
218
-
}
219
-
220
-
const obj = value as Record<string, unknown>;
221
-
222
-
return (
223
-
typeof obj.p === 'number' && isBytes(obj.k) && isCidLink(obj.v) && (obj.t === null || isCidLink(obj.t))
224
-
);
225
-
};
226
-
227
-
/** mst node object */
228
-
export interface MstNode {
229
-
/** link to sub-tree Node on a lower level and with all keys sorting before keys at this node */
230
-
l: CID.CidLink | null;
231
-
/** ordered list of TreeEntry objects */
232
-
e: TreeEntry[];
233
-
}
234
-
235
-
/**
236
-
* checks if a value is a valid mst node object
237
-
* @param value the value to check
238
-
* @returns true if the value is a valid mst node object, false otherwise
239
-
*/
240
-
export const isMstNode = (value: unknown): value is MstNode => {
241
-
if (value === null || typeof value !== 'object') {
242
-
return false;
243
-
}
244
-
245
-
const obj = value as Record<string, unknown>;
246
-
247
-
return (obj.l === null || isCidLink(obj.l)) && Array.isArray(obj.e) && obj.e.every(isTreeEntry);
248
-
};
packages/utilities/car/lib/index.ts
packages/utilities/car/lib/v3/index.ts
packages/utilities/car/lib/index.ts
packages/utilities/car/lib/v3/index.ts
+9
-6
packages/utilities/car/lib/reader.test.ts
packages/utilities/car/lib/v4/car-reader/index.test.ts
+9
-6
packages/utilities/car/lib/reader.test.ts
packages/utilities/car/lib/v4/car-reader/index.test.ts
···
3
3
import { fromString, toCidLink } from '@atcute/cid';
4
4
import { fromBase64 } from '@atcute/multibase';
5
5
6
-
import { readCar, readCarStream } from './reader.js';
6
+
import { fromStream } from './stream-car-reader.js';
7
+
import { fromUint8Array } from './sync-car-reader.js';
7
8
8
-
describe('readCar', () => {
9
+
describe('fromUint8Array', () => {
9
10
it('reads car files', () => {
10
11
const buf = fromBase64(
11
12
'OqJlcm9vdHOB2CpYJQABcRIgkD8I0DL+GsJ3OKREpf9k73yHguuSEYzEiXPGueoJg8FndmVy' +
···
29
30
'ZXQjbWVudGlvbmljcmVhdGVkQXR4GDIwMjQtMDItMjRUMTI6MTY6MjAuNjM3Wg',
30
31
);
31
32
32
-
const { header, iterate } = readCar(buf);
33
-
const entries = Array.from(iterate());
33
+
const car = fromUint8Array(buf);
34
+
35
+
const header = car.header;
36
+
const entries = Array.from(car);
34
37
35
38
expect(header).toEqual({
36
39
data: {
···
108
111
});
109
112
});
110
113
111
-
describe('readCarStream', () => {
114
+
describe('fromStream', () => {
112
115
it('reads car files', async () => {
113
116
const buf = fromBase64(
114
117
'OqJlcm9vdHOB2CpYJQABcRIgkD8I0DL+GsJ3OKREpf9k73yHguuSEYzEiXPGueoJg8FndmVy' +
···
135
138
const blob = new Blob([buf]);
136
139
const stream = blob.stream();
137
140
138
-
await using car = readCarStream(stream);
141
+
await using car = fromStream(stream);
139
142
140
143
const header = await car.header();
141
144
const entries = await Array.fromAsync(car);
-16
packages/utilities/car/lib/reader.ts
-16
packages/utilities/car/lib/reader.ts
···
1
-
import { createStreamedCarReader, type StreamedCarReader } from './utilities/stream-car-reader.js';
2
-
import { createUint8Reader } from './utilities/sync-byte-reader.js';
3
-
import { createCarReader, type SyncCarReader } from './utilities/sync-car-reader.js';
4
-
5
-
export type { CarEntry, CarHeader, CarV1Header } from './utilities/car.js';
6
-
export type { StreamedCarReader } from './utilities/stream-car-reader.js';
7
-
export type { SyncCarReader } from './utilities/sync-car-reader.js';
8
-
9
-
export const readCar = (buffer: Uint8Array): SyncCarReader => {
10
-
const reader = createUint8Reader(buffer);
11
-
return createCarReader(reader);
12
-
};
13
-
14
-
export const readCarStream = (stream: ReadableStream<Uint8Array>): StreamedCarReader => {
15
-
return createStreamedCarReader(stream);
16
-
};
packages/utilities/car/lib/utilities/car.ts
packages/utilities/car/lib/v4/car-reader/types.ts
packages/utilities/car/lib/utilities/car.ts
packages/utilities/car/lib/v4/car-reader/types.ts
+9
-3
packages/utilities/car/lib/utilities/stream-car-reader.ts
packages/utilities/car/lib/v4/car-reader/stream-car-reader.ts
+9
-3
packages/utilities/car/lib/utilities/stream-car-reader.ts
packages/utilities/car/lib/v4/car-reader/stream-car-reader.ts
···
2
2
import * as CID from '@atcute/cid';
3
3
import { concat } from '@atcute/uint8array';
4
4
5
-
import { isCarV1Header, type CarEntry, type CarHeader } from './car.js';
5
+
import { isCarV1Header, type CarEntry, type CarHeader } from './types.js';
6
6
7
7
export interface StreamedCarReader {
8
8
header(): Promise<CarHeader>;
9
9
roots(): Promise<CBOR.CidLink[]>;
10
10
11
+
dispose(): Promise<void>;
12
+
11
13
[Symbol.asyncDispose](): Promise<void>;
12
14
[Symbol.asyncIterator](): AsyncIterator<CarEntry>;
13
15
}
14
16
15
-
export const createStreamedCarReader = (stream: ReadableStream<Uint8Array>): StreamedCarReader => {
17
+
export const fromStream = (stream: ReadableStream<Uint8Array>): StreamedCarReader => {
16
18
let chunk = new Uint8Array(0) as Uint8Array; // annoying!
17
19
let offset = 0;
18
20
···
119
121
};
120
122
121
123
return {
122
-
async [Symbol.asyncDispose]() {
124
+
[Symbol.asyncDispose]() {
125
+
return this.dispose();
126
+
},
127
+
128
+
async dispose() {
123
129
await reader.cancel();
124
130
},
125
131
-39
packages/utilities/car/lib/utilities/sync-byte-reader.ts
-39
packages/utilities/car/lib/utilities/sync-byte-reader.ts
···
1
-
export interface SyncByteReader {
2
-
readonly pos: number;
3
-
upto(size: number): Uint8Array;
4
-
exactly(size: number, seek: boolean): Uint8Array;
5
-
seek(size: number): void;
6
-
}
7
-
8
-
export const createUint8Reader = (buf: Uint8Array): SyncByteReader => {
9
-
let pos = 0;
10
-
11
-
return {
12
-
get pos() {
13
-
return pos;
14
-
},
15
-
16
-
seek(size) {
17
-
if (size > buf.length - pos) {
18
-
throw new RangeError('unexpected end of data');
19
-
}
20
-
21
-
pos += size;
22
-
},
23
-
upto(size) {
24
-
return buf.subarray(pos, pos + size);
25
-
},
26
-
exactly(size, seek) {
27
-
if (size > buf.length - pos) {
28
-
throw new RangeError('unexpected end of data');
29
-
}
30
-
31
-
const slice = buf.subarray(pos, pos + size);
32
-
if (seek) {
33
-
pos += size;
34
-
}
35
-
36
-
return slice;
37
-
},
38
-
};
39
-
};
+90
-40
packages/utilities/car/lib/utilities/sync-car-reader.ts
packages/utilities/car/lib/v4/car-reader/sync-car-reader.ts
+90
-40
packages/utilities/car/lib/utilities/sync-car-reader.ts
packages/utilities/car/lib/v4/car-reader/sync-car-reader.ts
···
2
2
import * as CID from '@atcute/cid';
3
3
import * as varint from '@atcute/varint';
4
4
5
-
import { isCarV1Header, type CarEntry, type CarHeader } from './car.js';
6
-
import type { SyncByteReader } from './sync-byte-reader.js';
5
+
import { isCarV1Header, type CarEntry, type CarHeader } from './types.js';
6
+
7
+
interface SyncByteReader {
8
+
readonly pos: number;
9
+
upto(size: number): Uint8Array;
10
+
exactly(size: number, seek: boolean): Uint8Array;
11
+
seek(size: number): void;
12
+
}
7
13
8
14
export interface SyncCarReader {
9
-
header: CarHeader;
15
+
readonly header: CarHeader;
16
+
readonly roots: CBOR.CidLink[];
17
+
18
+
/** @deprecated do for..of on the reader directly */
10
19
iterate(): Generator<CarEntry>;
20
+
[Symbol.iterator](): Iterator<CarEntry>;
11
21
}
12
22
23
+
export const fromUint8Array = (buffer: Uint8Array): SyncCarReader => {
24
+
const reader = createUint8Reader(buffer);
25
+
const header = readHeader(reader);
26
+
27
+
return {
28
+
header,
29
+
roots: header.data.roots,
30
+
31
+
*iterate() {
32
+
while (reader.upto(8 + 36).length > 0) {
33
+
const entryStart = reader.pos;
34
+
const entrySize = readVarint(reader, 8);
35
+
36
+
const cidStart = reader.pos;
37
+
const cid = readCid(reader);
38
+
39
+
const bytesStart = reader.pos;
40
+
const bytesSize = entrySize - (bytesStart - cidStart);
41
+
const bytes = reader.exactly(bytesSize, true);
42
+
43
+
const cidEnd = bytesStart;
44
+
const bytesEnd = reader.pos;
45
+
const entryEnd = bytesEnd;
46
+
47
+
yield {
48
+
cid,
49
+
bytes,
50
+
51
+
entryStart,
52
+
entryEnd,
53
+
cidStart,
54
+
cidEnd,
55
+
bytesStart,
56
+
bytesEnd,
57
+
};
58
+
}
59
+
},
60
+
61
+
[Symbol.iterator](): Iterator<CarEntry> {
62
+
return this.iterate();
63
+
},
64
+
};
65
+
};
66
+
67
+
const createUint8Reader = (buf: Uint8Array): SyncByteReader => {
68
+
let pos = 0;
69
+
70
+
return {
71
+
get pos() {
72
+
return pos;
73
+
},
74
+
75
+
seek(size) {
76
+
if (size > buf.length - pos) {
77
+
throw new RangeError('unexpected end of data');
78
+
}
79
+
80
+
pos += size;
81
+
},
82
+
upto(size) {
83
+
return buf.subarray(pos, pos + size);
84
+
},
85
+
exactly(size, seek) {
86
+
if (size > buf.length - pos) {
87
+
throw new RangeError('unexpected end of data');
88
+
}
89
+
90
+
const slice = buf.subarray(pos, pos + size);
91
+
if (seek) {
92
+
pos += size;
93
+
}
94
+
95
+
return slice;
96
+
},
97
+
};
98
+
};
99
+
13
100
const readVarint = (reader: SyncByteReader, size: number): number => {
14
101
const buf = reader.upto(size);
15
102
if (buf.length === 0) {
···
82
169
83
170
return cid;
84
171
};
85
-
86
-
export const createCarReader = (reader: SyncByteReader): SyncCarReader => {
87
-
const header = readHeader(reader);
88
-
89
-
return {
90
-
header,
91
-
*iterate() {
92
-
while (reader.upto(8 + 36).length > 0) {
93
-
const entryStart = reader.pos;
94
-
const entrySize = readVarint(reader, 8);
95
-
96
-
const cidStart = reader.pos;
97
-
const cid = readCid(reader);
98
-
99
-
const bytesStart = reader.pos;
100
-
const bytesSize = entrySize - (bytesStart - cidStart);
101
-
const bytes = reader.exactly(bytesSize, true);
102
-
103
-
const cidEnd = bytesStart;
104
-
const bytesEnd = reader.pos;
105
-
const entryEnd = bytesEnd;
106
-
107
-
yield {
108
-
cid,
109
-
bytes,
110
-
111
-
entryStart,
112
-
entryEnd,
113
-
cidStart,
114
-
cidEnd,
115
-
bytesStart,
116
-
bytesEnd,
117
-
};
118
-
}
119
-
},
120
-
};
121
-
};
+15
packages/utilities/car/lib/v3/atproto-repo.ts
+15
packages/utilities/car/lib/v3/atproto-repo.ts
···
1
+
export {
2
+
collectBlock,
3
+
isCommit,
4
+
isMstNode,
5
+
isTreeEntry,
6
+
fromUint8Array as iterateAtpRepo,
7
+
readBlock,
8
+
RepoEntry,
9
+
walkMstEntries,
10
+
type BlockMap,
11
+
type Commit,
12
+
type MstNode,
13
+
type NodeEntry,
14
+
type TreeEntry,
15
+
} from '../v4/repo-reader/index.js';
+9
packages/utilities/car/lib/v3/reader.ts
+9
packages/utilities/car/lib/v3/reader.ts
+4
packages/utilities/car/lib/v4/car-reader/index.ts
+4
packages/utilities/car/lib/v4/car-reader/index.ts
+2
packages/utilities/car/lib/v4/index.ts
+2
packages/utilities/car/lib/v4/index.ts
+160
packages/utilities/car/lib/v4/repo-reader/index.test.ts
+160
packages/utilities/car/lib/v4/repo-reader/index.test.ts
···
1
+
import { describe, expect, it } from 'bun:test';
2
+
3
+
import { fromCidLink, toString } from '@atcute/cid';
4
+
import { fromBase64 } from '@atcute/multibase';
5
+
6
+
import { fromStream } from './stream-repo-reader.js';
7
+
import { fromUint8Array } from './sync-repo-reader.js';
8
+
9
+
describe('fromUint8Array', () => {
10
+
it('decodes atproto car files', () => {
11
+
const buf = fromBase64(
12
+
'OqJlcm9vdHOB2CpYJQABcRIgkD8I0DL+GsJ3OKREpf9k73yHguuSEYzEiXPGueoJg8FndmVy' +
13
+
'c2lvbgGPAQFxEiDqG8o/D37K3hldhQTMRq9/Uvyf7X9evn9eB9ZdgpYq6qRlJHR5cGV2YXBw' +
14
+
'LmJza3kuYWN0b3IucHJvZmlsZWljcmVhdGVkQXR4GDIwMjQtMDItMjRUMTI6MTU6NDEuMjE5' +
15
+
'WmtkZXNjcmlwdGlvbm90ZXN0aW5nIGFjY291bnRrZGlzcGxheU5hbWVg4AEBcRIgkD8I0DL+' +
16
+
'GsJ3OKREpf9k73yHguuSEYzEiXPGueoJg8GmY2RpZHggZGlkOnBsYzpzcmNxb3UybTd1cXVv' +
17
+
'Z3lkNXhrNGI1eTVjcmV2bTNsNXE1ZmplbnRjMmRjc2lnWEDeWWEO5/vV6SmnbUrLRu9WhWqI' +
18
+
'kHKANGFOin3xqFc4fgtuYzkbFXFJDMQU06nBWxict8FQ8Kas9Mr2fDAh++vVZGRhdGHYKlgl' +
19
+
'AAFxEiB2ibkpj3r4cdTag9v2ipIe8fxyjUFOgCjZbtYnfhyJ2GRwcmV29md2ZXJzaW9uA6QB' +
20
+
'AXESIHaJuSmPevhx1NqD2/aKkh7x/HKNQU6AKNlu1id+HInYomFlgaRha1gbYXBwLmJza3ku' +
21
+
'YWN0b3IucHJvZmlsZS9zZWxmYXAAYXTYKlglAAFxEiBvSJJSaF/w/fee+UmoLV84FDwZRC7p' +
22
+
'pJX484MghY0rM2F22CpYJQABcRIg6hvKPw9+yt4ZXYUEzEavf1L8n+1/Xr5/XgfWXYKWKuph' +
23
+
'bPaBAQFxEiBvSJJSaF/w/fee+UmoLV84FDwZRC7ppJX484MghY0rM6JhZYGkYWtYIGFwcC5i' +
24
+
'c2t5LmZlZWQucG9zdC8za201eW1rNGhoazJ6YXAAYXT2YXbYKlglAAFxEiDj+gU903L3F3Ar' +
25
+
'WCg+aeQZYEiM3ooIxqHbVvbQPZvEbGFs9qECAXESIOP6BT3TcvcXcCtYKD5p5BlgSIzeigjG' +
26
+
'odtW9tA9m8RspWR0ZXh0dWJlZXAgYm9vcCBAbWFyeS5teS5pZGUkdHlwZXJhcHAuYnNreS5m' +
27
+
'ZWVkLnBvc3RlbGFuZ3OBYmVuZmZhY2V0c4GjZSR0eXBld2FwcC5ic2t5LnJpY2h0ZXh0LmZh' +
28
+
'Y2V0ZWluZGV4omdieXRlRW5kFWlieXRlU3RhcnQKaGZlYXR1cmVzgaJjZGlkeCBkaWQ6cGxj' +
29
+
'OmlhNzZrdm5uZGp1dGdlZGdneDJpYnJlbWUkdHlwZXgfYXBwLmJza3kucmljaHRleHQuZmFj' +
30
+
'ZXQjbWVudGlvbmljcmVhdGVkQXR4GDIwMjQtMDItMjRUMTI6MTY6MjAuNjM3Wg',
31
+
);
32
+
33
+
const result = Array.from(fromUint8Array(buf), (entry) => ({
34
+
collection: entry.collection,
35
+
rkey: entry.rkey,
36
+
cid: toString(fromCidLink(entry.cid)),
37
+
record: entry.record,
38
+
}));
39
+
40
+
expect(result).toEqual([
41
+
{
42
+
collection: 'app.bsky.actor.profile',
43
+
rkey: 'self',
44
+
cid: 'bafyreihkdpfd6d36zlpbsxmfatgenl37kl6j73l7l27h6xqh2zoyffrk5i',
45
+
record: {
46
+
$type: 'app.bsky.actor.profile',
47
+
createdAt: '2024-02-24T12:15:41.219Z',
48
+
displayName: '',
49
+
description: 'testing account',
50
+
},
51
+
},
52
+
{
53
+
collection: 'app.bsky.feed.post',
54
+
rkey: '3km5ymk4hhk2z',
55
+
cid: 'bafyreihd7ict3u3s64lxak2yfa7gtzazmbeizxukbddkdw2w63id3g6enq',
56
+
record: {
57
+
$type: 'app.bsky.feed.post',
58
+
createdAt: '2024-02-24T12:16:20.637Z',
59
+
langs: ['en'],
60
+
text: 'beep boop @mary.my.id',
61
+
facets: [
62
+
{
63
+
$type: 'app.bsky.richtext.facet',
64
+
index: {
65
+
byteEnd: 21,
66
+
byteStart: 10,
67
+
},
68
+
features: [
69
+
{
70
+
did: 'did:plc:ia76kvnndjutgedggx2ibrem',
71
+
$type: 'app.bsky.richtext.facet#mention',
72
+
},
73
+
],
74
+
},
75
+
],
76
+
},
77
+
},
78
+
]);
79
+
});
80
+
});
81
+
82
+
describe('fromStream', () => {
83
+
it('decodes atproto car files', async () => {
84
+
const buf = fromBase64(
85
+
'OqJlcm9vdHOB2CpYJQABcRIgkD8I0DL+GsJ3OKREpf9k73yHguuSEYzEiXPGueoJg8FndmVy' +
86
+
'c2lvbgGPAQFxEiDqG8o/D37K3hldhQTMRq9/Uvyf7X9evn9eB9ZdgpYq6qRlJHR5cGV2YXBw' +
87
+
'LmJza3kuYWN0b3IucHJvZmlsZWljcmVhdGVkQXR4GDIwMjQtMDItMjRUMTI6MTU6NDEuMjE5' +
88
+
'WmtkZXNjcmlwdGlvbm90ZXN0aW5nIGFjY291bnRrZGlzcGxheU5hbWVg4AEBcRIgkD8I0DL+' +
89
+
'GsJ3OKREpf9k73yHguuSEYzEiXPGueoJg8GmY2RpZHggZGlkOnBsYzpzcmNxb3UybTd1cXVv' +
90
+
'Z3lkNXhrNGI1eTVjcmV2bTNsNXE1ZmplbnRjMmRjc2lnWEDeWWEO5/vV6SmnbUrLRu9WhWqI' +
91
+
'kHKANGFOin3xqFc4fgtuYzkbFXFJDMQU06nBWxict8FQ8Kas9Mr2fDAh++vVZGRhdGHYKlgl' +
92
+
'AAFxEiB2ibkpj3r4cdTag9v2ipIe8fxyjUFOgCjZbtYnfhyJ2GRwcmV29md2ZXJzaW9uA6QB' +
93
+
'AXESIHaJuSmPevhx1NqD2/aKkh7x/HKNQU6AKNlu1id+HInYomFlgaRha1gbYXBwLmJza3ku' +
94
+
'YWN0b3IucHJvZmlsZS9zZWxmYXAAYXTYKlglAAFxEiBvSJJSaF/w/fee+UmoLV84FDwZRC7p' +
95
+
'pJX484MghY0rM2F22CpYJQABcRIg6hvKPw9+yt4ZXYUEzEavf1L8n+1/Xr5/XgfWXYKWKuph' +
96
+
'bPaBAQFxEiBvSJJSaF/w/fee+UmoLV84FDwZRC7ppJX484MghY0rM6JhZYGkYWtYIGFwcC5i' +
97
+
'c2t5LmZlZWQucG9zdC8za201eW1rNGhoazJ6YXAAYXT2YXbYKlglAAFxEiDj+gU903L3F3Ar' +
98
+
'WCg+aeQZYEiM3ooIxqHbVvbQPZvEbGFs9qECAXESIOP6BT3TcvcXcCtYKD5p5BlgSIzeigjG' +
99
+
'odtW9tA9m8RspWR0ZXh0dWJlZXAgYm9vcCBAbWFyeS5teS5pZGUkdHlwZXJhcHAuYnNreS5m' +
100
+
'ZWVkLnBvc3RlbGFuZ3OBYmVuZmZhY2V0c4GjZSR0eXBld2FwcC5ic2t5LnJpY2h0ZXh0LmZh' +
101
+
'Y2V0ZWluZGV4omdieXRlRW5kFWlieXRlU3RhcnQKaGZlYXR1cmVzgaJjZGlkeCBkaWQ6cGxj' +
102
+
'OmlhNzZrdm5uZGp1dGdlZGdneDJpYnJlbWUkdHlwZXgfYXBwLmJza3kucmljaHRleHQuZmFj' +
103
+
'ZXQjbWVudGlvbmljcmVhdGVkQXR4GDIwMjQtMDItMjRUMTI6MTY6MjAuNjM3Wg',
104
+
);
105
+
106
+
const blob = new Blob([buf]);
107
+
const stream = blob.stream();
108
+
109
+
await using repo = fromStream(stream);
110
+
111
+
const result = await Array.fromAsync(repo, (entry) => ({
112
+
collection: entry.collection,
113
+
rkey: entry.rkey,
114
+
cid: toString(fromCidLink(entry.cid)),
115
+
record: entry.record,
116
+
}));
117
+
118
+
expect(repo.missingBlocks).toEqual([]);
119
+
120
+
expect(result).toEqual([
121
+
{
122
+
collection: 'app.bsky.actor.profile',
123
+
rkey: 'self',
124
+
cid: 'bafyreihkdpfd6d36zlpbsxmfatgenl37kl6j73l7l27h6xqh2zoyffrk5i',
125
+
record: {
126
+
$type: 'app.bsky.actor.profile',
127
+
createdAt: '2024-02-24T12:15:41.219Z',
128
+
displayName: '',
129
+
description: 'testing account',
130
+
},
131
+
},
132
+
{
133
+
collection: 'app.bsky.feed.post',
134
+
rkey: '3km5ymk4hhk2z',
135
+
cid: 'bafyreihd7ict3u3s64lxak2yfa7gtzazmbeizxukbddkdw2w63id3g6enq',
136
+
record: {
137
+
$type: 'app.bsky.feed.post',
138
+
createdAt: '2024-02-24T12:16:20.637Z',
139
+
langs: ['en'],
140
+
text: 'beep boop @mary.my.id',
141
+
facets: [
142
+
{
143
+
$type: 'app.bsky.richtext.facet',
144
+
index: {
145
+
byteEnd: 21,
146
+
byteStart: 10,
147
+
},
148
+
features: [
149
+
{
150
+
did: 'did:plc:ia76kvnndjutgedggx2ibrem',
151
+
$type: 'app.bsky.richtext.facet#mention',
152
+
},
153
+
],
154
+
},
155
+
],
156
+
},
157
+
},
158
+
]);
159
+
});
160
+
});
+5
packages/utilities/car/lib/v4/repo-reader/index.ts
+5
packages/utilities/car/lib/v4/repo-reader/index.ts
+110
packages/utilities/car/lib/v4/repo-reader/mst.ts
+110
packages/utilities/car/lib/v4/repo-reader/mst.ts
···
1
+
import * as CBOR from '@atcute/cbor';
2
+
import * as CID from '@atcute/cid';
3
+
4
+
const isCidLink = (value: unknown): value is CID.CidLink => {
5
+
if (value instanceof CID.CidLinkWrapper) {
6
+
return true;
7
+
}
8
+
9
+
if (value === null || typeof value !== 'object') {
10
+
return false;
11
+
}
12
+
13
+
return '$link' in value && typeof value.$link === 'string';
14
+
};
15
+
16
+
const isBytes = (value: unknown): value is CBOR.Bytes => {
17
+
if (value instanceof CBOR.BytesWrapper) {
18
+
return true;
19
+
}
20
+
21
+
if (value === null || typeof value !== 'object') {
22
+
return false;
23
+
}
24
+
25
+
return '$bytes' in value && typeof value.$bytes === 'string';
26
+
};
27
+
28
+
/** commit object */
29
+
export interface Commit {
30
+
version: 3;
31
+
did: string;
32
+
data: CID.CidLink;
33
+
rev: string;
34
+
prev: CID.CidLink | null;
35
+
sig: CBOR.Bytes;
36
+
}
37
+
38
+
/**
39
+
* checks if a value is a valid commit object
40
+
* @param value the value to check
41
+
* @returns true if the value is a valid commit object, false otherwise
42
+
*/
43
+
export const isCommit = (value: unknown): value is Commit => {
44
+
if (value === null || typeof value !== 'object') {
45
+
return false;
46
+
}
47
+
48
+
const obj = value as Record<string, unknown>;
49
+
50
+
return (
51
+
obj.version === 3 &&
52
+
typeof obj.did === 'string' &&
53
+
isCidLink(obj.data) &&
54
+
typeof obj.rev === 'string' &&
55
+
(obj.prev === null || isCidLink(obj.prev)) &&
56
+
isBytes(obj.sig)
57
+
);
58
+
};
59
+
60
+
/** mst tree entry object */
61
+
export interface TreeEntry {
62
+
/** count of bytes shared with previous TreeEntry in this Node (if any) */
63
+
p: number;
64
+
/** remainder of key for this TreeEntry, after "prefixlen" have been removed */
65
+
k: CBOR.Bytes;
66
+
/** link to a sub-tree Node at a lower level which has keys sorting after this TreeEntry's key (to the "right"), but before the next TreeEntry's key in this Node (if any) */
67
+
v: CID.CidLink;
68
+
/** next subtree (to the right of leaf) */
69
+
t: CID.CidLink | null;
70
+
}
71
+
72
+
/**
73
+
* checks if a value is a valid mst tree entry object
74
+
* @param value the value to check
75
+
* @returns true if the value is a valid mst tree entry object, false otherwise
76
+
*/
77
+
export const isTreeEntry = (value: unknown): value is TreeEntry => {
78
+
if (value === null || typeof value !== 'object') {
79
+
return false;
80
+
}
81
+
82
+
const obj = value as Record<string, unknown>;
83
+
84
+
return (
85
+
typeof obj.p === 'number' && isBytes(obj.k) && isCidLink(obj.v) && (obj.t === null || isCidLink(obj.t))
86
+
);
87
+
};
88
+
89
+
/** mst node object */
90
+
export interface MstNode {
91
+
/** link to sub-tree Node on a lower level and with all keys sorting before keys at this node */
92
+
l: CID.CidLink | null;
93
+
/** ordered list of TreeEntry objects */
94
+
e: TreeEntry[];
95
+
}
96
+
97
+
/**
98
+
* checks if a value is a valid mst node object
99
+
* @param value the value to check
100
+
* @returns true if the value is a valid mst node object, false otherwise
101
+
*/
102
+
export const isMstNode = (value: unknown): value is MstNode => {
103
+
if (value === null || typeof value !== 'object') {
104
+
return false;
105
+
}
106
+
107
+
const obj = value as Record<string, unknown>;
108
+
109
+
return (obj.l === null || isCidLink(obj.l)) && Array.isArray(obj.e) && obj.e.every(isTreeEntry);
110
+
};
+173
packages/utilities/car/lib/v4/repo-reader/stream-repo-reader.ts
+173
packages/utilities/car/lib/v4/repo-reader/stream-repo-reader.ts
···
1
+
import Queue from 'yocto-queue';
2
+
3
+
import * as CBOR from '@atcute/cbor';
4
+
import * as CID from '@atcute/cid';
5
+
import { decodeUtf8From } from '@atcute/uint8array';
6
+
7
+
import * as CarReader from '../car-reader/index.js';
8
+
import { assert } from '../utils.js';
9
+
10
+
import { isCommit, isMstNode } from './mst.js';
11
+
import { RepoEntry } from './types.js';
12
+
13
+
type EntryMeta = { t: 0 } | { t: 1 } | { t: 2; k: string };
14
+
15
+
type Task = {
16
+
c: string;
17
+
e: CarReader.CarEntry;
18
+
m: EntryMeta;
19
+
};
20
+
21
+
export type MissingBlockEntry =
22
+
| {
23
+
cid: string;
24
+
type: 'record';
25
+
key: string;
26
+
}
27
+
| {
28
+
cid: string;
29
+
type: 'mst-node';
30
+
}
31
+
| {
32
+
cid: string;
33
+
type: 'commit';
34
+
};
35
+
36
+
export interface StreamedRepoReader {
37
+
/**
38
+
* list of blocks that were referenced but not found in the repository.
39
+
* blocks may be reported as missing if multiple records share the same CID.
40
+
*/
41
+
readonly missingBlocks: readonly MissingBlockEntry[];
42
+
43
+
dispose(): Promise<void>;
44
+
45
+
[Symbol.asyncDispose](): Promise<void>;
46
+
[Symbol.asyncIterator](): AsyncIterator<RepoEntry>;
47
+
}
48
+
49
+
export const fromStream = (stream: ReadableStream<Uint8Array>): StreamedRepoReader => {
50
+
let missingBlocks: MissingBlockEntry[] = [];
51
+
52
+
return {
53
+
get missingBlocks() {
54
+
return missingBlocks;
55
+
},
56
+
57
+
async dispose() {
58
+
// does nothing for now
59
+
},
60
+
61
+
[Symbol.asyncDispose]() {
62
+
return this.dispose();
63
+
},
64
+
async *[Symbol.asyncIterator]() {
65
+
await using car = CarReader.fromStream(stream);
66
+
67
+
const pending = new Map<string, EntryMeta>();
68
+
const strays = new Map<string, CarReader.CarEntry>();
69
+
70
+
const queue = new Queue<Task>();
71
+
72
+
const request = (cid: string, meta: EntryMeta): void => {
73
+
const entry = strays.get(cid);
74
+
75
+
if (entry !== undefined) {
76
+
strays.delete(cid);
77
+
queue.enqueue({ c: cid, e: entry, m: meta });
78
+
} else {
79
+
pending.set(cid, meta);
80
+
}
81
+
};
82
+
83
+
{
84
+
const roots = await car.roots();
85
+
assert(roots.length === 1, `expected only 1 root in the car archive; got=${roots.length}`);
86
+
87
+
const rootCid = roots[0].$link;
88
+
request(rootCid, { t: 0 });
89
+
}
90
+
91
+
for await (const entry of car) {
92
+
const cid = CID.toString(entry.cid);
93
+
94
+
{
95
+
const meta = pending.get(cid);
96
+
97
+
if (meta !== undefined) {
98
+
pending.delete(cid);
99
+
queue.enqueue({ c: cid, e: entry, m: meta });
100
+
} else {
101
+
strays.set(cid, entry);
102
+
}
103
+
}
104
+
105
+
let task: Task | undefined;
106
+
while ((task = queue.dequeue())) {
107
+
const { c: cid, e: entry, m: meta } = task;
108
+
109
+
switch (meta.t) {
110
+
case 0: {
111
+
const commit = CBOR.decode(entry.bytes);
112
+
assert(isCommit(commit), `expected commit block; cid=${cid}`);
113
+
114
+
request(commit.data.$link, { t: 1 });
115
+
break;
116
+
}
117
+
case 1: {
118
+
const node = CBOR.decode(entry.bytes);
119
+
assert(isMstNode(node), `expected mst node block; cid=${cid}`);
120
+
121
+
const entries = node.e;
122
+
const left = node.l;
123
+
124
+
let lastKey = '';
125
+
126
+
if (left !== null) {
127
+
request(left.$link, meta);
128
+
}
129
+
130
+
for (let i = 0, il = entries.length; i < il; i++) {
131
+
const entry = entries[i];
132
+
const next = entry.t;
133
+
134
+
const key_str = decodeUtf8From(CBOR.fromBytes(entry.k));
135
+
const key = lastKey.slice(0, entry.p) + key_str;
136
+
137
+
lastKey = key;
138
+
139
+
request(entry.v.$link, { t: 2, k: key });
140
+
141
+
if (next !== null) {
142
+
request(next.$link, { t: 1 });
143
+
}
144
+
}
145
+
146
+
break;
147
+
}
148
+
case 2: {
149
+
const [collection, rkey] = meta.k.split('/');
150
+
151
+
yield new RepoEntry(collection, rkey, CID.toCidLink(entry.cid), entry);
152
+
break;
153
+
}
154
+
}
155
+
}
156
+
}
157
+
158
+
missingBlocks = Array.from(pending, ([cid, meta]): MissingBlockEntry => {
159
+
switch (meta.t) {
160
+
case 0: {
161
+
return { cid, type: 'commit' };
162
+
}
163
+
case 1: {
164
+
return { cid, type: 'mst-node' };
165
+
}
166
+
case 2: {
167
+
return { cid, type: 'record', key: meta.k };
168
+
}
169
+
}
170
+
});
171
+
},
172
+
};
173
+
};
+85
packages/utilities/car/lib/v4/repo-reader/sync-blockmap.ts
+85
packages/utilities/car/lib/v4/repo-reader/sync-blockmap.ts
···
1
+
import * as CBOR from '@atcute/cbor';
2
+
import * as CID from '@atcute/cid';
3
+
import { decodeUtf8From } from '@atcute/uint8array';
4
+
5
+
import * as CarReader from '../car-reader/index.js';
6
+
import { assert } from '../utils.js';
7
+
8
+
import { isMstNode } from './mst.js';
9
+
10
+
export type BlockMap = Map<string, CarReader.CarEntry>;
11
+
12
+
/**
13
+
* collects entries from a CAR archive into a mapping of CID string -> actual bytes
14
+
* @param iterator a generator that yields objects with a `cid` and `bytes` property
15
+
* @returns a mapping of CID string -> actual bytes
16
+
*/
17
+
export const collectBlock = (iterator: Iterable<CarReader.CarEntry>): BlockMap => {
18
+
const blockmap: BlockMap = new Map();
19
+
for (const entry of iterator) {
20
+
blockmap.set(CID.toString(entry.cid), entry);
21
+
}
22
+
23
+
return blockmap;
24
+
};
25
+
26
+
/**
27
+
* reads a block from the blockmap and validates it against the provided validation function
28
+
* @param map a mapping of CID string -> actual bytes
29
+
* @param link a CID link to read
30
+
* @param validate a validation function to validate the decoded data
31
+
* @returns the decoded and validated data
32
+
*/
33
+
export const readBlock = <T>(
34
+
map: BlockMap,
35
+
link: CID.CidLink,
36
+
validate: (value: unknown) => value is T,
37
+
): T => {
38
+
const cid = link.$link;
39
+
40
+
const entry = map.get(cid);
41
+
assert(entry != null, `cid not found in blockmap; cid=${cid}`);
42
+
43
+
const data = CBOR.decode(entry.bytes);
44
+
assert(validate(data), `validation failed for cid=${cid}`);
45
+
46
+
return data;
47
+
};
48
+
49
+
/** node entry object */
50
+
export interface NodeEntry {
51
+
key: string;
52
+
cid: CID.CidLink;
53
+
}
54
+
55
+
/**
56
+
* walks the entries of a Merkle Sorted Tree (MST) in a depth-first manner
57
+
* @param map a mapping of CID string -> actual bytes
58
+
* @param pointer a CID link to the root of the MST
59
+
* @returns a generator that yields the entries of the MST
60
+
*/
61
+
export function* walkMstEntries(map: BlockMap, pointer: CID.CidLink): Generator<NodeEntry> {
62
+
const data = readBlock(map, pointer, isMstNode);
63
+
const entries = data.e;
64
+
65
+
let lastKey = '';
66
+
67
+
if (data.l !== null) {
68
+
yield* walkMstEntries(map, data.l);
69
+
}
70
+
71
+
for (let i = 0, il = entries.length; i < il; i++) {
72
+
const entry = entries[i];
73
+
74
+
const key_str = decodeUtf8From(CBOR.fromBytes(entry.k));
75
+
const key = lastKey.slice(0, entry.p) + key_str;
76
+
77
+
lastKey = key;
78
+
79
+
yield { key: key, cid: entry.v };
80
+
81
+
if (entry.t !== null) {
82
+
yield* walkMstEntries(map, entry.t);
83
+
}
84
+
}
85
+
}
+27
packages/utilities/car/lib/v4/repo-reader/sync-repo-reader.ts
+27
packages/utilities/car/lib/v4/repo-reader/sync-repo-reader.ts
···
1
+
import * as CarReader from '../car-reader/index.js';
2
+
import { assert } from '../utils.js';
3
+
4
+
import { isCommit } from './mst.js';
5
+
import { collectBlock, readBlock, walkMstEntries } from './sync-blockmap.js';
6
+
import { RepoEntry } from './types.js';
7
+
8
+
export function* fromUint8Array(buf: Uint8Array): Generator<RepoEntry> {
9
+
const car = CarReader.fromUint8Array(buf);
10
+
const roots = car.roots;
11
+
12
+
assert(roots.length === 1, `expected only 1 root in the car archive; got=${roots.length}`);
13
+
14
+
const blockmap = collectBlock(car);
15
+
assert(blockmap.size > 0, `expected at least 1 block in the archive; got=${blockmap.size}`);
16
+
17
+
const commit = readBlock(blockmap, roots[0], isCommit);
18
+
19
+
for (const { key, cid } of walkMstEntries(blockmap, commit.data)) {
20
+
const [collection, rkey] = key.split('/');
21
+
22
+
const carEntry = blockmap.get(cid.$link);
23
+
assert(carEntry != null, `cid not found in blockmap; cid=${cid}`);
24
+
25
+
yield new RepoEntry(collection, rkey, cid, carEntry);
26
+
}
27
+
}
+32
packages/utilities/car/lib/v4/repo-reader/types.ts
+32
packages/utilities/car/lib/v4/repo-reader/types.ts
···
1
+
import * as CBOR from '@atcute/cbor';
2
+
import * as CID from '@atcute/cid';
3
+
4
+
import { type CarEntry } from '../car-reader/index.js';
5
+
6
+
export class RepoEntry {
7
+
/** @internal */
8
+
constructor(
9
+
/** the collection this record belongs to */
10
+
public readonly collection: string,
11
+
/** record key */
12
+
public readonly rkey: string,
13
+
/** CID of this record */
14
+
public readonly cid: CID.CidLink,
15
+
/** the associated CarEntry for this record */
16
+
public readonly carEntry: CarEntry,
17
+
) {}
18
+
19
+
/**
20
+
* raw contents of this record
21
+
*/
22
+
get bytes(): Uint8Array {
23
+
return this.carEntry.bytes;
24
+
}
25
+
26
+
/**
27
+
* decoded contents of this record
28
+
*/
29
+
get record(): unknown {
30
+
return CBOR.decode(this.bytes);
31
+
}
32
+
}
+7
packages/utilities/car/lib/v4/utils.ts
+7
packages/utilities/car/lib/v4/utils.ts
+6
-2
packages/utilities/car/package.json
+6
-2
packages/utilities/car/package.json
···
20
20
"!lib/**/*.test.ts"
21
21
],
22
22
"exports": {
23
-
".": "./dist/index.js"
23
+
".": "./dist/v3/index.js",
24
+
"./v4": "./dist/v4/index.js",
25
+
"./v4/car-reader": "./dist/v4/car-reader/index.js",
26
+
"./v4/repo-reader": "./dist/v4/repo-reader/index.js"
24
27
},
25
28
"sideEffects": false,
26
29
"scripts": {
···
36
39
"@atcute/cbor": "workspace:^",
37
40
"@atcute/cid": "workspace:^",
38
41
"@atcute/uint8array": "workspace:^",
39
-
"@atcute/varint": "workspace:^"
42
+
"@atcute/varint": "workspace:^",
43
+
"yocto-queue": "^1.2.1"
40
44
}
41
45
}
+1
packages/utilities/car/tsconfig.json
+1
packages/utilities/car/tsconfig.json