handy online tools for AT Protocol
boat.kelinci.net
atproto
bluesky
atcute
typescript
solidjs
feat: blob export
+364
-1
+50
src/lib/utils/promise-queue.ts
+50
src/lib/utils/promise-queue.ts
···
1
+
export class PromiseQueue {
2
+
#queue: { deferred: PromiseWithResolvers<any>; fn: () => any }[] = [];
3
+
4
+
#max: number;
5
+
#current = 0;
6
+
7
+
constructor({ max = 2 }: { max?: number } = {}) {
8
+
this.#max = max;
9
+
}
10
+
11
+
add<T>(fn: () => Promise<T>): Promise<T> {
12
+
const deferred = Promise.withResolvers<T>();
13
+
14
+
this.#queue.push({ deferred, fn });
15
+
this.#run();
16
+
17
+
return deferred.promise;
18
+
}
19
+
20
+
async flush(): Promise<void> {
21
+
while (this.#queue.length > 0) {
22
+
await Promise.all(this.#queue.map((task) => task.deferred.promise));
23
+
}
24
+
}
25
+
26
+
#run() {
27
+
if (this.#queue.length > 0 && this.#current <= this.#max) {
28
+
const { deferred, fn } = this.#queue.shift()!;
29
+
this.#current++;
30
+
31
+
const promise = new Promise((r) => r(fn()));
32
+
33
+
const done = () => {
34
+
this.#current--;
35
+
this.#run();
36
+
};
37
+
38
+
promise.then(
39
+
(res) => {
40
+
done();
41
+
deferred.resolve(res);
42
+
},
43
+
(err) => {
44
+
done();
45
+
deferred.reject(err);
46
+
},
47
+
);
48
+
}
49
+
}
50
+
}
+5
src/routes.ts
+5
src/routes.ts
+308
src/views/blob/blob-export.tsx
+308
src/views/blob/blob-export.tsx
···
1
+
import { FileSystemWritableFileStream, showSaveFilePicker } from 'native-file-system-adapter';
2
+
import { createSignal } from 'solid-js';
3
+
import * as v from 'valibot';
4
+
5
+
import { simpleFetchHandler, XRPC, XRPCError } from '@atcute/client';
6
+
import { At } from '@atcute/client/lexicons';
7
+
import { writeTarEntry } from '@mary/tar';
8
+
9
+
import { getDidDocument } from '~/api/queries/did-doc';
10
+
import { resolveHandleViaAppView, resolveHandleViaPds } from '~/api/queries/handle';
11
+
import { getPdsEndpoint } from '~/api/types/did-doc';
12
+
import { serviceUrlString } from '~/api/types/strings';
13
+
import { DID_OR_HANDLE_RE, isDid } from '~/api/utils/strings';
14
+
15
+
import { makeAbortable } from '~/lib/utils/abortable';
16
+
import { PromiseQueue } from '~/lib/utils/promise-queue';
17
+
18
+
import Logger, { createLogger } from '~/components/logger';
19
+
20
+
const BlobExportPage = () => {
21
+
const logger = createLogger();
22
+
23
+
const [getSignal, cleanup] = makeAbortable();
24
+
const [pending, setPending] = createSignal(false);
25
+
26
+
const mutate = async ({
27
+
identifier,
28
+
service,
29
+
signal,
30
+
}: {
31
+
identifier: string;
32
+
service?: string;
33
+
signal?: AbortSignal;
34
+
}) => {
35
+
logger.info(`Starting export for ${identifier}`);
36
+
37
+
let did: At.DID;
38
+
if (isDid(identifier)) {
39
+
did = identifier;
40
+
} else if (service) {
41
+
did = await resolveHandleViaPds({ service, handle: identifier, signal });
42
+
logger.log(`Resolved handle to ${did}`);
43
+
} else {
44
+
did = await resolveHandleViaAppView({ handle: identifier, signal });
45
+
logger.log(`Resolved handle to ${did}`);
46
+
}
47
+
48
+
if (!service) {
49
+
const didDoc = await getDidDocument({ did, signal });
50
+
logger.log(`Retrieved DID document`);
51
+
52
+
const endpoint = getPdsEndpoint(didDoc);
53
+
if (!endpoint) {
54
+
logger.error(`Identity does not have a PDS server set`);
55
+
return;
56
+
}
57
+
58
+
logger.log(`PDS located at ${endpoint}`);
59
+
service = endpoint;
60
+
}
61
+
62
+
const rpc = new XRPC({ handler: simpleFetchHandler({ service }) });
63
+
64
+
// Grab a list of blobs
65
+
let blobs: string[] = [];
66
+
{
67
+
using progress = logger.progress(`Retrieving list of blobs`);
68
+
69
+
let cursor: string | undefined;
70
+
do {
71
+
const { data } = await rpc.get('com.atproto.sync.listBlobs', {
72
+
signal,
73
+
params: { did, cursor, limit: 1_000 },
74
+
});
75
+
76
+
cursor = data.cursor;
77
+
blobs = blobs.concat(data.cids);
78
+
79
+
progress.update(`Retrieving list of blobs (found ${blobs.length})`);
80
+
} while (cursor !== undefined);
81
+
82
+
logger.log(`Found ${blobs.length} blobs to download`);
83
+
if (blobs.length === 0) {
84
+
logger.warn(`Nothing to do`);
85
+
return;
86
+
}
87
+
}
88
+
89
+
// Now ask the user to save
90
+
let writable: FileSystemWritableFileStream | undefined;
91
+
{
92
+
using _progress = logger.progress(`Waiting for the user`);
93
+
94
+
const fd = await showSaveFilePicker({
95
+
suggestedName: `blobs-${identifier}-${new Date().toISOString()}.tar`,
96
+
97
+
// @ts-expect-error: ponyfill doesn't have the full typings
98
+
id: 'blob-export',
99
+
startIn: 'downloads',
100
+
types: [
101
+
{
102
+
description: 'Tarball archive',
103
+
accept: { 'application/tar': ['.tar'] },
104
+
},
105
+
],
106
+
}).catch((err) => {
107
+
console.warn(err);
108
+
109
+
if (err instanceof DOMException && err.name === 'AbortError') {
110
+
logger.warn(`Opened the file picker, but it was aborted`);
111
+
} else {
112
+
logger.warn(`Something went wrong when opening the file picker`);
113
+
}
114
+
115
+
return undefined;
116
+
});
117
+
118
+
writable = await fd?.createWritable();
119
+
120
+
if (writable === undefined) {
121
+
// We already handled the errors above
122
+
return;
123
+
}
124
+
125
+
signal?.throwIfAborted();
126
+
}
127
+
128
+
// Let's download!
129
+
{
130
+
let downloadedCount = 0;
131
+
132
+
using progress = logger.progress(`Downloading blobs (${downloadedCount} of ${blobs.length})`);
133
+
134
+
const queue = new PromiseQueue();
135
+
for (const cid of blobs) {
136
+
queue.add(async () => {
137
+
const download = async () => {
138
+
let attempts = 0;
139
+
140
+
while (true) {
141
+
if (attempts > 0) {
142
+
await sleep(2_000);
143
+
}
144
+
145
+
attempts++;
146
+
147
+
try {
148
+
const { data } = await rpc.get('com.atproto.sync.getBlob', {
149
+
signal,
150
+
params: { did, cid },
151
+
});
152
+
153
+
return data;
154
+
} catch (err) {
155
+
if (attempts > 3) {
156
+
throw err;
157
+
}
158
+
159
+
if (err instanceof XRPCError) {
160
+
if (err.status === 400) {
161
+
if (err.message === 'Blob not found') {
162
+
console.warn(`Blob ${cid} not found`);
163
+
return;
164
+
}
165
+
} else if (err.status === 429) {
166
+
const reset = err.headers?.['ratelimit-reset'];
167
+
168
+
if (reset !== undefined) {
169
+
logger.warn(`Ratelimit exceeded when downloading ${cid}, waiting`);
170
+
171
+
const refreshAt = +reset * 1_000;
172
+
const delta = refreshAt - Date.now();
173
+
174
+
await sleep(delta);
175
+
}
176
+
}
177
+
}
178
+
}
179
+
}
180
+
};
181
+
182
+
const data = await download();
183
+
if (data !== undefined) {
184
+
const entry = writeTarEntry({ filename: `blobs/${cid}`, data });
185
+
writable.write(entry);
186
+
}
187
+
188
+
progress.update(`Downloading blobs (${++downloadedCount} of ${blobs.length})`);
189
+
});
190
+
}
191
+
192
+
// Await for everything here.
193
+
await queue.flush();
194
+
}
195
+
196
+
// We're done here.
197
+
{
198
+
using _progress = logger.progress(`Flushing writes`);
199
+
await writable.close();
200
+
}
201
+
202
+
logger.log(`Finished!`);
203
+
};
204
+
205
+
return (
206
+
<>
207
+
<div class="p-4">
208
+
<h1 class="text-lg font-bold text-purple-800">Export blobs</h1>
209
+
<p class="text-gray-600">Download all blobs from an account</p>
210
+
</div>
211
+
<hr class="mx-4 border-gray-300" />
212
+
213
+
<form
214
+
onSubmit={(ev) => {
215
+
const formEl = ev.currentTarget;
216
+
const formData = new FormData(formEl);
217
+
ev.preventDefault();
218
+
219
+
const signal = getSignal();
220
+
221
+
const ident = formData.get('ident') as string;
222
+
const service = formData.get('service') as string;
223
+
224
+
const promise = mutate({
225
+
identifier: ident,
226
+
service: service || undefined,
227
+
signal,
228
+
});
229
+
230
+
setPending(true);
231
+
232
+
promise.then(
233
+
() => {
234
+
if (signal.aborted) {
235
+
return;
236
+
}
237
+
238
+
cleanup();
239
+
setPending(false);
240
+
},
241
+
(err) => {
242
+
if (signal.aborted) {
243
+
return;
244
+
}
245
+
246
+
cleanup();
247
+
setPending(false);
248
+
logger.error(`Critical error: ${err}`);
249
+
},
250
+
);
251
+
}}
252
+
class="m-4 flex flex-col gap-4"
253
+
>
254
+
<fieldset disabled={pending()} class="contents">
255
+
<label class="flex flex-col gap-2">
256
+
<span class="font-semibold text-gray-600">Handle or DID identifier*</span>
257
+
<input
258
+
type="text"
259
+
name="ident"
260
+
required
261
+
pattern={DID_OR_HANDLE_RE.source}
262
+
placeholder="paul.bsky.social"
263
+
class="rounded border border-gray-400 px-3 py-2 text-sm outline-2 -outline-offset-1 outline-purple-600 placeholder:text-gray-400 focus:outline"
264
+
/>
265
+
</label>
266
+
267
+
<label class="flex flex-col gap-2">
268
+
<span class="font-semibold text-gray-600">PDS service</span>
269
+
<input
270
+
type="url"
271
+
name="service"
272
+
placeholder="https://bsky.social"
273
+
onInput={(ev) => {
274
+
const input = ev.currentTarget;
275
+
const value = input.value;
276
+
277
+
if (value !== '' && !v.is(serviceUrlString, value)) {
278
+
input.setCustomValidity('Must be a valid service URL');
279
+
} else {
280
+
input.setCustomValidity('');
281
+
}
282
+
}}
283
+
class="rounded border border-gray-400 px-3 py-2 text-sm outline-2 -outline-offset-1 outline-purple-600 placeholder:text-gray-400 focus:outline"
284
+
/>
285
+
</label>
286
+
287
+
<div>
288
+
<button
289
+
type="submit"
290
+
class="flex h-9 select-none items-center rounded bg-purple-800 px-4 text-sm font-semibold text-white hover:bg-purple-700 active:bg-purple-700 disabled:pointer-events-none disabled:opacity-50"
291
+
>
292
+
Export!
293
+
</button>
294
+
</div>
295
+
</fieldset>
296
+
</form>
297
+
<hr class="mx-4 border-gray-300" />
298
+
299
+
<Logger logger={logger} />
300
+
</>
301
+
);
302
+
};
303
+
304
+
export default BlobExportPage;
305
+
306
+
const sleep = (ms: number): Promise<void> => {
307
+
return new Promise((resolve) => setTimeout(resolve, ms));
308
+
};