Something went wrong. Try again.
A nicer S3 CLI
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214import * as FS from 'node:fs';import * as PATH from 'node:path';import { pipeline } from 'node:stream/promises';import { Transform } from 'node:stream';import type { Bucket } from './remote.ts';import { setting } from './config.ts';import { progress } from './output.ts';import { MAX_SINGLE_COPY, type ListEntry, type ObjectEntry } from './vendor/ordner.fyi/s3/s3.ts';
// Moving bytes: object to file, file to object, object to object — all// streamed, so a file of any size passes through a bounded amount of memory,// with a progress line on a terminal. Uploads larger than one part go up in// parts, several at once; the vendored client does that, and aborts the// upload at the service if it fails or is stopped.
export interface Transfer { signal?: AbortSignal | undefined; /** Show a progress line (a terminal, and not --json). */ show: boolean;}
const MiB = 1024 * 1024;
export function partSize(): number { return Number(setting('part-size')) * MiB;}
export function concurrency(): number { return Number(setting('concurrency'));}
/** * A file to write a download into. Its `fs` has no writev: inside the bundled * archive (node:vfs, mounted by the launcher), WriteStream's writev to a real * file fails with "h.writev is not a function", and without one node writes * each chunk with write instead. */export function fileSink(path: string): FS.WriteStream { return FS.createWriteStream(path, { fs: { open: FS.open, write: FS.write, close: FS.close } } as Parameters<typeof FS.createWriteStream>[1]);}
/** Every entry directly in a folder: its objects and its subfolders. */export async function listFolder(bucket: Bucket, prefix: string, signal?: AbortSignal): Promise<ListEntry[]> { return await bucket.run(async (client) => { const entries: ListEntry[] = []; for await (const entry of client.list(prefix, { delimiter: '/', signal })) entries.push(entry); return entries; });}
/** Every object beneath a prefix, however deep. */export async function listAll(bucket: Bucket, prefix: string, signal?: AbortSignal): Promise<ObjectEntry[]> { return await bucket.run(async (client) => { const entries: ObjectEntry[] = []; for await (const entry of client.list(prefix, { signal })) entries.push(entry); return entries; });}
/** Whether anything at all is beneath a prefix: one page is asked for, and only the first entry read. */export async function hasAny(bucket: Bucket, prefix: string, signal?: AbortSignal): Promise<boolean> { return await bucket.run(async (client) => { for await (const _ of client.list(prefix, { delimiter: '/', signal })) return true; return false; });}
function counter(bar: ReturnType<typeof progress>) { return new Transform({ transform(chunk: Buffer, _encoding, done) { bar.add(chunk.length); done(null, chunk); }, });}
/** An object into a file, through a temporary file beside it, so a cut-off download leaves no half file. */export async function download(bucket: Bucket, key: string, file: string, options: Transfer): Promise<number> { FS.mkdirSync(PATH.dirname(file), { recursive: true }); const temp = `${file}.s3cli-${process.pid}.part`; const { body, size } = await bucket.run((client) => client.stream(key, { signal: options.signal })); const bar = progress(key, size, options.show); try { await pipeline(body, counter(bar), fileSink(temp), { signal: options.signal }); FS.renameSync(temp, file); } catch (error) { FS.rmSync(temp, { force: true }); throw error; } finally { bar.done(); } return size ?? FS.statSync(file).size;}
/** An object onto a stream (stdout, for cat and `get … -`). */export async function pipeTo(bucket: Bucket, key: string, out: NodeJS.WritableStream, signal?: AbortSignal): Promise<void> { const { body } = await bucket.run((client) => client.stream(key, { signal })); await pipeline(body, out, { signal, end: false });}
/** * A stream up as one object: read until it is more than a part, and if it * ended first, sent whole in one PUT; else in parts, starting with what was * read. So `put - key` from a small pipe is one request, and from a large one * never more than a few parts in memory. */export async function uploadStream(bucket: Bucket, key: string, source: AsyncIterable<Buffer>, options: Transfer & { contentType?: string | undefined; size?: number | undefined; label?: string }): Promise<number> { const part = partSize(); const bar = progress(options.label ?? key, options.size, options.show); const iterator = source[Symbol.asyncIterator](); const head: Buffer[] = []; let total = 0; let ended = false; try { while (total <= part) { const next = await iterator.next(); if (next.done) { ended = true; break; } head.push(next.value); total += next.value.length; } if (ended) { const body = Buffer.concat(head); await bucket.run((client) => client.put(key, body, { contentType: options.contentType, signal: options.signal })); bar.add(body.length); return body.length; } bar.add(total); async function* rest(): AsyncGenerator<Buffer> { yield* head; for (;;) { const next = await iterator.next(); if (next.done) return; total += next.value.length; bar.add(next.value.length); yield next.value; } } // Not through bucket.run: a stream cannot be read twice. Its region is known by now, from listing or a first call. await bucket.client.put(key, rest(), { contentType: options.contentType, partSize: part, partConcurrency: concurrency(), signal: options.signal }); return total; } finally { bar.done(); }}
/** A file up as one object. */export async function upload(bucket: Bucket, file: string, key: string, options: Transfer & { contentType?: string | undefined }): Promise<number> { const size = FS.statSync(file).size; const contentType = options.contentType ?? contentTypeOf(file); if (size <= partSize()) { const body = FS.readFileSync(file); const bar = progress(key, size, options.show); try { await bucket.run((client) => client.put(key, body, { contentType, signal: options.signal })); bar.add(size); } finally { bar.done(); } return size; } return await uploadStream(bucket, key, FS.createReadStream(file, { highWaterMark: MiB }) as AsyncIterable<Buffer>, { ...options, contentType, size });}
/** * An object to another place: copied at the service when both are in one * bucket, else streamed through here (another bucket, another service). */export async function copyObject(from: Bucket, fromKey: string, to: Bucket, toKey: string, size: number, options: Transfer): Promise<'server' | 'streamed'> { if (from.alias === to.alias && from.profile.name === to.profile.name) { await from.run((client) => size > MAX_SINGLE_COPY ? client.copyMultipart(toKey, fromKey, { size, signal: options.signal }) : client.copy(toKey, fromKey, { signal: options.signal })); return 'server'; } const { body, type } = await from.run((client) => client.stream(fromKey, { signal: options.signal })); await uploadStream(to, toKey, body, { ...options, contentType: type, size, label: `${fromKey} → ${toKey}` }); return 'streamed';}
/** Run `fn` over `items`, at most `limit` at once; the first failure stops what has not started. */export async function eachLimit<T>(items: readonly T[], limit: number, fn: (item: T) => Promise<void>): Promise<void> { let next = 0; let failure: unknown; const worker = async () => { while (failure === undefined && next < items.length) { const item = items[next++]!; try { await fn(item); } catch (error) { failure ??= error; } } }; await Promise.all(Array.from({ length: Math.min(limit, items.length) }, worker)); if (failure !== undefined) throw failure;}
const TYPES: Record<string, string> = { txt: 'text/plain; charset=utf-8', md: 'text/markdown; charset=utf-8', csv: 'text/csv; charset=utf-8', log: 'text/plain; charset=utf-8', html: 'text/html; charset=utf-8', htm: 'text/html; charset=utf-8', css: 'text/css; charset=utf-8', js: 'text/javascript; charset=utf-8', mjs: 'text/javascript; charset=utf-8', json: 'application/json', xml: 'application/xml', yaml: 'application/yaml', yml: 'application/yaml', pdf: 'application/pdf', zip: 'application/zip', gz: 'application/gzip', tar: 'application/x-tar', wasm: 'application/wasm', svg: 'image/svg+xml', png: 'image/png', jpg: 'image/jpeg', jpeg: 'image/jpeg', gif: 'image/gif', webp: 'image/webp', avif: 'image/avif', ico: 'image/vnd.microsoft.icon', mp3: 'audio/mpeg', ogg: 'audio/ogg', wav: 'audio/wav', mp4: 'video/mp4', webm: 'video/webm', mov: 'video/quicktime', woff: 'font/woff', woff2: 'font/woff2', ttf: 'font/ttf', otf: 'font/otf',};
/** A content type from a file name's extension; application/octet-stream when it says nothing. */export function contentTypeOf(name: string): string { return TYPES[PATH.extname(name).slice(1).toLowerCase()] ?? 'application/octet-stream';}