a collection of lightweight TypeScript packages for AT Protocol, the protocol powering Bluesky
atproto
bluesky
typescript
npm
feat(car): `repoEntryTransform` and `carEntryTransform` functions
closes https://github.com/mary-ext/atcute/pull/34
closes https://github.com/mary-ext/atcute/issues/33
Co-authored-by: Mary <git@mary.my.id>
authored by
Nick the Sick
Mary
and committed by
Mary
074a818b
150edc27
+255
-2
+20
.changeset/bumpy-maps-care.md
+20
.changeset/bumpy-maps-care.md
···
1
+
---
2
+
'@atcute/car': minor
3
+
---
4
+
5
+
adds `RepoReader.repoEntryTransform` and `CarReader.carEntryTransform` functions for conveniently
6
+
piping a readable stream into archive entries.
7
+
8
+
```ts
9
+
import { RepoReader } from '@atcute/car/v4';
10
+
11
+
const response = await fetch('https://example.com/xrpc/com.atproto.sync.getRepo?did=...');
12
+
13
+
const entries = response.body!.pipeThrough(RepoReader.repoEntryTransform());
14
+
for await (const entry of entries) {
15
+
entry;
16
+
// ^? RepoEntry { ... }
17
+
}
18
+
```
19
+
20
+
thanks [@nperez0111](https://github.com/nperez0111) for this contribution!
+95
-1
packages/utilities/car/lib/v4/car-reader/index.test.ts
+95
-1
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 { fromStream } from './stream-car-reader.js';
6
+
import { carEntryTransform, fromStream } from './stream-car-reader.js';
7
7
import { fromUint8Array } from './sync-car-reader.js';
8
8
9
9
describe('fromUint8Array', () => {
···
218
218
]);
219
219
});
220
220
});
221
+
222
+
describe('carEntryTransform', () => {
223
+
it('reads car files', async () => {
224
+
const buf = fromBase64(
225
+
'OqJlcm9vdHOB2CpYJQABcRIgkD8I0DL+GsJ3OKREpf9k73yHguuSEYzEiXPGueoJg8FndmVy' +
226
+
'c2lvbgGPAQFxEiDqG8o/D37K3hldhQTMRq9/Uvyf7X9evn9eB9ZdgpYq6qRlJHR5cGV2YXBw' +
227
+
'LmJza3kuYWN0b3IucHJvZmlsZWljcmVhdGVkQXR4GDIwMjQtMDItMjRUMTI6MTU6NDEuMjE5' +
228
+
'WmtkZXNjcmlwdGlvbm90ZXN0aW5nIGFjY291bnRrZGlzcGxheU5hbWVg4AEBcRIgkD8I0DL+' +
229
+
'GsJ3OKREpf9k73yHguuSEYzEiXPGueoJg8GmY2RpZHggZGlkOnBsYzpzcmNxb3UybTd1cXVv' +
230
+
'Z3lkNXhrNGI1eTVjcmV2bTNsNXE1ZmplbnRjMmRjc2lnWEDeWWEO5/vV6SmnbUrLRu9WhWqI' +
231
+
'kHKANGFOin3xqFc4fgtuYzkbFXFJDMQU06nBWxict8FQ8Kas9Mr2fDAh++vVZGRhdGHYKlgl' +
232
+
'AAFxEiB2ibkpj3r4cdTag9v2ipIe8fxyjUFOgCjZbtYnfhyJ2GRwcmV29md2ZXJzaW9uA6QB' +
233
+
'AXESIHaJuSmPevhx1NqD2/aKkh7x/HKNQU6AKNlu1id+HInYomFlgaRha1gbYXBwLmJza3ku' +
234
+
'YWN0b3IucHJvZmlsZS9zZWxmYXAAYXTYKlglAAFxEiBvSJJSaF/w/fee+UmoLV84FDwZRC7p' +
235
+
'pJX484MghY0rM2F22CpYJQABcRIg6hvKPw9+yt4ZXYUEzEavf1L8n+1/Xr5/XgfWXYKWKuph' +
236
+
'bPaBAQFxEiBvSJJSaF/w/fee+UmoLV84FDwZRC7ppJX484MghY0rM6JhZYGkYWtYIGFwcC5i' +
237
+
'c2t5LmZlZWQucG9zdC8za201eW1rNGhoazJ6YXAAYXT2YXbYKlglAAFxEiDj+gU903L3F3Ar' +
238
+
'WCg+aeQZYEiM3ooIxqHbVvbQPZvEbGFs9qECAXESIOP6BT3TcvcXcCtYKD5p5BlgSIzeigjG' +
239
+
'odtW9tA9m8RspWR0ZXh0dWJlZXAgYm9vcCBAbWFyeS5teS5pZGUkdHlwZXJhcHAuYnNreS5m' +
240
+
'ZWVkLnBvc3RlbGFuZ3OBYmVuZmZhY2V0c4GjZSR0eXBld2FwcC5ic2t5LnJpY2h0ZXh0LmZh' +
241
+
'Y2V0ZWluZGV4omdieXRlRW5kFWlieXRlU3RhcnQKaGZlYXR1cmVzgaJjZGlkeCBkaWQ6cGxj' +
242
+
'OmlhNzZrdm5uZGp1dGdlZGdneDJpYnJlbWUkdHlwZXgfYXBwLmJza3kucmljaHRleHQuZmFj' +
243
+
'ZXQjbWVudGlvbmljcmVhdGVkQXR4GDIwMjQtMDItMjRUMTI6MTY6MjAuNjM3Wg',
244
+
);
245
+
246
+
const blob = new Blob([buf]);
247
+
const stream = blob.stream().pipeThrough(carEntryTransform());
248
+
249
+
const entries = await Array.fromAsync(stream);
250
+
251
+
expect(entries).toEqual([
252
+
{
253
+
cid: fromString('bafyreihkdpfd6d36zlpbsxmfatgenl37kl6j73l7l27h6xqh2zoyffrk5i'),
254
+
bytes: fromBase64(
255
+
'pGUkdHlwZXZhcHAuYnNreS5hY3Rvci5wcm9maWxlaWNyZWF0ZWRBdHgYMjAyNC0wMi0yNFQxMjoxNTo0MS4yMTlaa2Rlc2NyaXB0aW9ub3Rlc3RpbmcgYWNjb3VudGtkaXNwbGF5TmFtZWA',
256
+
),
257
+
entryEnd: 204,
258
+
entryStart: 59,
259
+
cidEnd: 97,
260
+
cidStart: 61,
261
+
bytesEnd: 204,
262
+
bytesStart: 97,
263
+
},
264
+
{
265
+
cid: fromString('bafyreieqh4enamx6dlbhoofeiss76zhppsdyf24scggmjclty246ucmdye'),
266
+
bytes: fromBase64(
267
+
'pmNkaWR4IGRpZDpwbGM6c3JjcW91Mm03dXF1b2d5ZDV4azRiNXk1Y3Jldm0zbDVxNWZqZW50YzJkY3NpZ1hA3llhDuf71ekpp21Ky0bvVoVqiJBygDRhTop98ahXOH4LbmM5GxVxSQzEFNOpwVsYnLfBUPCmrPTK9nwwIfvr1WRkYXRh2CpYJQABcRIgdom5KY96+HHU2oPb9oqSHvH8co1BToAo2W7WJ34cidhkcHJldvZndmVyc2lvbgM',
268
+
),
269
+
entryEnd: 430,
270
+
entryStart: 204,
271
+
cidEnd: 242,
272
+
cidStart: 206,
273
+
bytesEnd: 430,
274
+
bytesStart: 242,
275
+
},
276
+
{
277
+
cid: fromString('bafyreidwrg4std327by5jwud3p3iveq66h6hfdkbj2acrwlo2ytx4hej3a'),
278
+
bytes: fromBase64(
279
+
'omFlgaRha1gbYXBwLmJza3kuYWN0b3IucHJvZmlsZS9zZWxmYXAAYXTYKlglAAFxEiBvSJJSaF/w/fee+UmoLV84FDwZRC7ppJX484MghY0rM2F22CpYJQABcRIg6hvKPw9+yt4ZXYUEzEavf1L8n+1/Xr5/XgfWXYKWKuphbPY',
280
+
),
281
+
entryEnd: 596,
282
+
entryStart: 430,
283
+
cidEnd: 468,
284
+
cidStart: 432,
285
+
bytesEnd: 596,
286
+
bytesStart: 468,
287
+
},
288
+
{
289
+
cid: fromString('bafyreidpjcjfe2c76d67phxzjguc2xzycq6bsrbo5gsjl6htqmqildjlgm'),
290
+
bytes: fromBase64(
291
+
'omFlgaRha1ggYXBwLmJza3kuZmVlZC5wb3N0LzNrbTV5bWs0aGhrMnphcABhdPZhdtgqWCUAAXESIOP6BT3TcvcXcCtYKD5p5BlgSIzeigjGodtW9tA9m8RsYWz2',
292
+
),
293
+
entryEnd: 727,
294
+
entryStart: 596,
295
+
cidEnd: 634,
296
+
cidStart: 598,
297
+
bytesEnd: 727,
298
+
bytesStart: 634,
299
+
},
300
+
{
301
+
cid: fromString('bafyreihd7ict3u3s64lxak2yfa7gtzazmbeizxukbddkdw2w63id3g6enq'),
302
+
bytes: fromBase64(
303
+
'pWR0ZXh0dWJlZXAgYm9vcCBAbWFyeS5teS5pZGUkdHlwZXJhcHAuYnNreS5mZWVkLnBvc3RlbGFuZ3OBYmVuZmZhY2V0c4GjZSR0eXBld2FwcC5ic2t5LnJpY2h0ZXh0LmZhY2V0ZWluZGV4omdieXRlRW5kFWlieXRlU3RhcnQKaGZlYXR1cmVzgaJjZGlkeCBkaWQ6cGxjOmlhNzZrdm5uZGp1dGdlZGdneDJpYnJlbWUkdHlwZXgfYXBwLmJza3kucmljaHRleHQuZmFjZXQjbWVudGlvbmljcmVhdGVkQXR4GDIwMjQtMDItMjRUMTI6MTY6MjAuNjM3Wg',
304
+
),
305
+
entryEnd: 1018,
306
+
entryStart: 727,
307
+
cidEnd: 765,
308
+
cidStart: 729,
309
+
bytesEnd: 1018,
310
+
bytesStart: 765,
311
+
},
312
+
]);
313
+
});
314
+
});
+31
packages/utilities/car/lib/v4/car-reader/stream-car-reader.ts
+31
packages/utilities/car/lib/v4/car-reader/stream-car-reader.ts
···
14
14
[Symbol.asyncIterator](): AsyncIterator<CarEntry>;
15
15
}
16
16
17
+
export const carEntryTransform = (): ReadableWritablePair<CarEntry, Uint8Array> => {
18
+
const transform = new TransformStream<Uint8Array, Uint8Array>();
19
+
let car: StreamedCarReader | undefined;
20
+
21
+
return {
22
+
readable: new ReadableStream({
23
+
async start(controller) {
24
+
car = fromStream(transform.readable);
25
+
26
+
try {
27
+
for await (const entry of car) {
28
+
controller.enqueue(entry);
29
+
}
30
+
31
+
await car.dispose();
32
+
33
+
controller.close();
34
+
} catch (err) {
35
+
controller.error(err);
36
+
}
37
+
},
38
+
async cancel() {
39
+
if (car !== undefined) {
40
+
await car.dispose();
41
+
}
42
+
},
43
+
}),
44
+
writable: transform.writable,
45
+
};
46
+
};
47
+
17
48
export const fromStream = (stream: ReadableStream<Uint8Array>): StreamedCarReader => {
18
49
let chunk = new Uint8Array(0) as Uint8Array; // annoying!
19
50
let offset = 0;
+77
-1
packages/utilities/car/lib/v4/repo-reader/index.test.ts
+77
-1
packages/utilities/car/lib/v4/repo-reader/index.test.ts
···
3
3
import { fromCidLink, toString } from '@atcute/cid';
4
4
import { fromBase64 } from '@atcute/multibase';
5
5
6
-
import { fromStream } from './stream-repo-reader.js';
6
+
import { fromStream, repoEntryTransform } from './stream-repo-reader.js';
7
7
import { fromUint8Array } from './sync-repo-reader.js';
8
8
9
9
describe('fromUint8Array', () => {
···
158
158
]);
159
159
});
160
160
});
161
+
162
+
describe('repoEntryTransform', () => {
163
+
it('decodes atproto car files', async () => {
164
+
const buf = fromBase64(
165
+
'OqJlcm9vdHOB2CpYJQABcRIgkD8I0DL+GsJ3OKREpf9k73yHguuSEYzEiXPGueoJg8FndmVy' +
166
+
'c2lvbgGPAQFxEiDqG8o/D37K3hldhQTMRq9/Uvyf7X9evn9eB9ZdgpYq6qRlJHR5cGV2YXBw' +
167
+
'LmJza3kuYWN0b3IucHJvZmlsZWljcmVhdGVkQXR4GDIwMjQtMDItMjRUMTI6MTU6NDEuMjE5' +
168
+
'WmtkZXNjcmlwdGlvbm90ZXN0aW5nIGFjY291bnRrZGlzcGxheU5hbWVg4AEBcRIgkD8I0DL+' +
169
+
'GsJ3OKREpf9k73yHguuSEYzEiXPGueoJg8GmY2RpZHggZGlkOnBsYzpzcmNxb3UybTd1cXVv' +
170
+
'Z3lkNXhrNGI1eTVjcmV2bTNsNXE1ZmplbnRjMmRjc2lnWEDeWWEO5/vV6SmnbUrLRu9WhWqI' +
171
+
'kHKANGFOin3xqFc4fgtuYzkbFXFJDMQU06nBWxict8FQ8Kas9Mr2fDAh++vVZGRhdGHYKlgl' +
172
+
'AAFxEiB2ibkpj3r4cdTag9v2ipIe8fxyjUFOgCjZbtYnfhyJ2GRwcmV29md2ZXJzaW9uA6QB' +
173
+
'AXESIHaJuSmPevhx1NqD2/aKkh7x/HKNQU6AKNlu1id+HInYomFlgaRha1gbYXBwLmJza3ku' +
174
+
'YWN0b3IucHJvZmlsZS9zZWxmYXAAYXTYKlglAAFxEiBvSJJSaF/w/fee+UmoLV84FDwZRC7p' +
175
+
'pJX484MghY0rM2F22CpYJQABcRIg6hvKPw9+yt4ZXYUEzEavf1L8n+1/Xr5/XgfWXYKWKuph' +
176
+
'bPaBAQFxEiBvSJJSaF/w/fee+UmoLV84FDwZRC7ppJX484MghY0rM6JhZYGkYWtYIGFwcC5i' +
177
+
'c2t5LmZlZWQucG9zdC8za201eW1rNGhoazJ6YXAAYXT2YXbYKlglAAFxEiDj+gU903L3F3Ar' +
178
+
'WCg+aeQZYEiM3ooIxqHbVvbQPZvEbGFs9qECAXESIOP6BT3TcvcXcCtYKD5p5BlgSIzeigjG' +
179
+
'odtW9tA9m8RspWR0ZXh0dWJlZXAgYm9vcCBAbWFyeS5teS5pZGUkdHlwZXJhcHAuYnNreS5m' +
180
+
'ZWVkLnBvc3RlbGFuZ3OBYmVuZmZhY2V0c4GjZSR0eXBld2FwcC5ic2t5LnJpY2h0ZXh0LmZh' +
181
+
'Y2V0ZWluZGV4omdieXRlRW5kFWlieXRlU3RhcnQKaGZlYXR1cmVzgaJjZGlkeCBkaWQ6cGxj' +
182
+
'OmlhNzZrdm5uZGp1dGdlZGdneDJpYnJlbWUkdHlwZXgfYXBwLmJza3kucmljaHRleHQuZmFj' +
183
+
'ZXQjbWVudGlvbmljcmVhdGVkQXR4GDIwMjQtMDItMjRUMTI6MTY6MjAuNjM3Wg',
184
+
);
185
+
186
+
const blob = new Blob([buf]);
187
+
const stream = blob.stream().pipeThrough(repoEntryTransform());
188
+
189
+
const result = await Array.fromAsync(stream, (entry) => ({
190
+
collection: entry.collection,
191
+
rkey: entry.rkey,
192
+
cid: toString(fromCidLink(entry.cid)),
193
+
record: entry.record,
194
+
}));
195
+
196
+
expect(result).toEqual([
197
+
{
198
+
collection: 'app.bsky.actor.profile',
199
+
rkey: 'self',
200
+
cid: 'bafyreihkdpfd6d36zlpbsxmfatgenl37kl6j73l7l27h6xqh2zoyffrk5i',
201
+
record: {
202
+
$type: 'app.bsky.actor.profile',
203
+
createdAt: '2024-02-24T12:15:41.219Z',
204
+
displayName: '',
205
+
description: 'testing account',
206
+
},
207
+
},
208
+
{
209
+
collection: 'app.bsky.feed.post',
210
+
rkey: '3km5ymk4hhk2z',
211
+
cid: 'bafyreihd7ict3u3s64lxak2yfa7gtzazmbeizxukbddkdw2w63id3g6enq',
212
+
record: {
213
+
$type: 'app.bsky.feed.post',
214
+
createdAt: '2024-02-24T12:16:20.637Z',
215
+
langs: ['en'],
216
+
text: 'beep boop @mary.my.id',
217
+
facets: [
218
+
{
219
+
$type: 'app.bsky.richtext.facet',
220
+
index: {
221
+
byteEnd: 21,
222
+
byteStart: 10,
223
+
},
224
+
features: [
225
+
{
226
+
did: 'did:plc:ia76kvnndjutgedggx2ibrem',
227
+
$type: 'app.bsky.richtext.facet#mention',
228
+
},
229
+
],
230
+
},
231
+
],
232
+
},
233
+
},
234
+
]);
235
+
});
236
+
});
+1
packages/utilities/car/lib/v4/repo-reader/index.ts
+1
packages/utilities/car/lib/v4/repo-reader/index.ts
+31
packages/utilities/car/lib/v4/repo-reader/stream-repo-reader.ts
+31
packages/utilities/car/lib/v4/repo-reader/stream-repo-reader.ts
···
46
46
[Symbol.asyncIterator](): AsyncIterator<RepoEntry>;
47
47
}
48
48
49
+
export const repoEntryTransform = (): ReadableWritablePair<RepoEntry, Uint8Array> => {
50
+
const transform = new TransformStream<Uint8Array, Uint8Array>();
51
+
let repo: StreamedRepoReader | undefined;
52
+
53
+
return {
54
+
readable: new ReadableStream({
55
+
async start(controller) {
56
+
repo = fromStream(transform.readable);
57
+
58
+
try {
59
+
for await (const entry of repo) {
60
+
controller.enqueue(entry);
61
+
}
62
+
63
+
await repo.dispose();
64
+
65
+
controller.close();
66
+
} catch (err) {
67
+
controller.error(err);
68
+
}
69
+
},
70
+
async cancel() {
71
+
if (repo !== undefined) {
72
+
await repo.dispose();
73
+
}
74
+
},
75
+
}),
76
+
writable: transform.writable,
77
+
};
78
+
};
79
+
49
80
export const fromStream = (stream: ReadableStream<Uint8Array>): StreamedRepoReader => {
50
81
let missingBlocks: MissingBlockEntry[] = [];
51
82