chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
// @pdsjs/blobs-s3 - S3-compatible blob storage (AWS S3, MinIO, R2 via S3 API)
export { createS3FetchBackupTarget } from './fetch-target.js';
/**
* @typedef {import('@pdsjs/core/ports').BlobPort} BlobPort
* @typedef {import('@pdsjs/core/ports').BackupTargetPort} BackupTargetPort
*/
/**
* @typedef {Object} S3Client
* @property {(command: any) => Promise<any>} send - Send a command to S3
*/
// Multipart part size. S3 requires at least 5 MiB for every part but the
// last.
const S3_PART_SIZE = 8 * 1024 * 1024;
/**
* @param {Uint8Array[]} chunks
* @param {number} size
* @returns {Uint8Array}
*/
function joinChunks(chunks, size) {
const out = new Uint8Array(size);
let at = 0;
for (const chunk of chunks) {
out.set(chunk, at);
at += chunk.length;
}
return out;
}
/**
* Create S3-compatible blob adapter
* Works with AWS S3, MinIO, Cloudflare R2 (via S3 API), and other S3-compatible storage
*
* The five multipart and copy constructors are optional as one group. With
* them, the adapter offers `createWriter` and a streamed upload stages under
* a temporary key, lands under its CID through a server-side copy, and holds
* one part buffer in memory at most. Without them, core buffers streamed
* uploads and calls `put`.
* @param {Object} options
* @param {S3Client} options.client - S3 client instance
* @param {string} options.bucket - Bucket name
* @param {new (params: any) => any} options.GetObjectCommand - GetObjectCommand constructor
* @param {new (params: any) => any} options.PutObjectCommand - PutObjectCommand constructor
* @param {new (params: any) => any} options.DeleteObjectCommand - DeleteObjectCommand constructor
* @param {new (params: any) => any} [options.CreateMultipartUploadCommand]
* @param {new (params: any) => any} [options.UploadPartCommand]
* @param {new (params: any) => any} [options.CompleteMultipartUploadCommand]
* @param {new (params: any) => any} [options.AbortMultipartUploadCommand]
* @param {new (params: any) => any} [options.CopyObjectCommand]
* @returns {BlobPort}
*/
export function createS3Blobs({
client,
bucket,
GetObjectCommand,
PutObjectCommand,
DeleteObjectCommand,
CreateMultipartUploadCommand,
UploadPartCommand,
CompleteMultipartUploadCommand,
AbortMultipartUploadCommand,
CopyObjectCommand,
}) {
const multipartReady = Boolean(
CreateMultipartUploadCommand &&
UploadPartCommand &&
CompleteMultipartUploadCommand &&
AbortMultipartUploadCommand &&
CopyObjectCommand,
);
/**
* @param {string} did
* @returns {Promise<import('@pdsjs/core/ports').BlobWriter>}
*/
async function createWriter(did) {
const tempKey = `${did}/tmp/${crypto.randomUUID()}`;
/** @type {string|null} */
let uploadId = null;
/** @type {{ETag: string, PartNumber: number}[]} */
const parts = [];
/** @type {Uint8Array[]} */
let pending = [];
let pendingSize = 0;
async function flushPart() {
if (!uploadId) {
const created = await client.send(
new /** @type {new (params: any) => any} */ (
CreateMultipartUploadCommand
)({ Bucket: bucket, Key: tempKey }),
);
uploadId = created.UploadId;
}
const body = joinChunks(pending, pendingSize);
const sent = await client.send(
new /** @type {new (params: any) => any} */ (UploadPartCommand)({
Bucket: bucket,
Key: tempKey,
UploadId: uploadId,
PartNumber: parts.length + 1,
Body: body,
}),
);
parts.push({ ETag: sent.ETag, PartNumber: parts.length + 1 });
pending = [];
pendingSize = 0;
}
return {
/** @param {Uint8Array} chunk */
async write(chunk) {
pending.push(chunk);
pendingSize += chunk.length;
if (pendingSize >= S3_PART_SIZE) await flushPart();
},
/**
* @param {string} cid
* @param {string} mimeType
*/
async commit(cid, mimeType) {
const finalKey = `${did}/${cid}`;
if (!uploadId) {
await client.send(
new PutObjectCommand({
Bucket: bucket,
Key: finalKey,
Body: joinChunks(pending, pendingSize),
ContentType: mimeType,
Metadata: { mime_type: mimeType },
}),
);
return;
}
if (pendingSize > 0) await flushPart();
await client.send(
new /** @type {new (params: any) => any} */ (
CompleteMultipartUploadCommand
)({
Bucket: bucket,
Key: tempKey,
UploadId: uploadId,
MultipartUpload: { Parts: parts },
}),
);
await client.send(
new /** @type {new (params: any) => any} */ (CopyObjectCommand)({
Bucket: bucket,
Key: finalKey,
CopySource: encodeURI(`${bucket}/${tempKey}`),
MetadataDirective: 'REPLACE',
ContentType: mimeType,
Metadata: { mime_type: mimeType },
}),
);
await client.send(
new DeleteObjectCommand({ Bucket: bucket, Key: tempKey }),
);
},
async abort() {
if (uploadId) {
await client.send(
new /** @type {new (params: any) => any} */ (
AbortMultipartUploadCommand
)({ Bucket: bucket, Key: tempKey, UploadId: uploadId }),
);
}
pending = [];
pendingSize = 0;
},
};
}
return {
...(multipartReady ? { createWriter } : {}),
/**
* @param {string} did
* @param {string} cid
* @param {{start: number, end: number}} [range] - Inclusive byte positions
*/
async getStream(did, cid, range) {
try {
const response = await client.send(
new GetObjectCommand({
Bucket: bucket,
Key: `${did}/${cid}`,
...(range ? { Range: `bytes=${range.start}-${range.end}` } : {}),
}),
);
return {
body: /** @type {ReadableStream<Uint8Array>} */ (
response.Body.transformToWebStream()
),
mimeType:
response.Metadata?.mime_type ||
response.ContentType ||
'application/octet-stream',
};
} catch (e) {
if (e.name === 'NoSuchKey') return null;
throw e;
}
},
/**
* @param {string} did
* @param {string} cid
*/
async get(did, cid) {
try {
const response = await client.send(
new GetObjectCommand({
Bucket: bucket,
Key: `${did}/${cid}`,
}),
);
const data = new Uint8Array(await response.Body.transformToByteArray());
const mimeType =
response.Metadata?.mime_type || 'application/octet-stream';
return { data, mimeType };
} catch (e) {
if (e.name === 'NoSuchKey') return null;
throw e;
}
},
/**
* @param {string} did
* @param {string} cid
* @param {Uint8Array} data
* @param {string} mimeType
*/
async put(did, cid, data, mimeType) {
await client.send(
new PutObjectCommand({
Bucket: bucket,
Key: `${did}/${cid}`,
Body: data,
ContentType: mimeType,
Metadata: { mime_type: mimeType },
}),
);
},
/**
* @param {string} did
* @param {string} cid
*/
async delete(did, cid) {
await client.send(
new DeleteObjectCommand({
Bucket: bucket,
Key: `${did}/${cid}`,
}),
);
},
};
}
/**
* Backups in an S3-compatible bucket, following the injected-client pattern
* above so the AWS SDK stays out of this package's dependencies. Give
* backups their own bucket: core deletes anything under `snapshots/` and
* `blobs/` that its retention policy no longer wants.
* @param {Object} options
* @param {S3Client} options.client - S3 client instance
* @param {string} options.bucket - Bucket name
* @param {new (params: any) => any} options.GetObjectCommand
* @param {new (params: any) => any} options.PutObjectCommand
* @param {new (params: any) => any} options.DeleteObjectCommand
* @param {new (params: any) => any} options.ListObjectsV2Command
* @returns {BackupTargetPort}
*/
export function createS3BackupTarget({
client,
bucket,
GetObjectCommand,
PutObjectCommand,
DeleteObjectCommand,
ListObjectsV2Command,
}) {
return {
/**
* @param {string} key
* @param {Uint8Array} data
* @param {string} [contentType]
*/
async put(key, data, contentType) {
await client.send(
new PutObjectCommand({
Bucket: bucket,
Key: key,
Body: data,
ContentType: contentType || 'application/octet-stream',
}),
);
},
/** @param {string} key */
async get(key) {
try {
const response = await client.send(
new GetObjectCommand({ Bucket: bucket, Key: key }),
);
return new Uint8Array(await response.Body.transformToByteArray());
} catch (e) {
if (e.name === 'NoSuchKey' || e.name === 'NotFound') return null;
throw e;
}
},
/** @param {string} prefix */
async list(prefix) {
/** @type {string[]} */
const keys = [];
/** @type {string|undefined} */
let continuationToken;
do {
const response = await client.send(
new ListObjectsV2Command({
Bucket: bucket,
Prefix: prefix,
ContinuationToken: continuationToken,
}),
);
for (const object of response.Contents || []) keys.push(object.Key);
continuationToken = response.IsTruncated
? response.NextContinuationToken
: undefined;
} while (continuationToken);
return keys;
},
/** @param {string} key */
async delete(key) {
await client.send(new DeleteObjectCommand({ Bucket: bucket, Key: key }));
},
};
}