// @pdsjs/blobs-fs - Filesystem blob storage import { randomUUID } from 'node:crypto'; import { createReadStream } from 'node:fs'; import { mkdir, open, readdir, readFile, rename, unlink, writeFile, } from 'node:fs/promises'; import { dirname, join, relative, sep } from 'node:path'; import { Readable } from 'node:stream'; /** * @typedef {import('@pdsjs/core/ports').BlobPort} BlobPort * @typedef {import('@pdsjs/core/ports').BackupTargetPort} BackupTargetPort */ /** * Create filesystem blob adapter * @param {string} baseDir - Base directory for blob storage * @returns {BlobPort} */ export function createFsBlobs(baseDir) { /** * @param {string} did * @param {string} cid * @returns {string} */ function blobPath(did, cid) { // Use first 2 chars of CID for sharding const shard = cid.slice(1, 3); return join(baseDir, did, shard, cid); } return { /** * @param {string} did * @param {string} cid */ async get(did, cid) { try { const path = blobPath(did, cid); const data = await readFile(path); const metaPath = `${path}.meta`; const meta = JSON.parse(await readFile(metaPath, 'utf8')); return { data: new Uint8Array(data), mimeType: meta.mimeType }; } catch (e) { if (/** @type {NodeJS.ErrnoException} */ (e).code === 'ENOENT') return null; throw e; } }, /** * @param {string} did * @param {string} cid * @param {Uint8Array} data * @param {string} mimeType */ async put(did, cid, data, mimeType) { const path = blobPath(did, cid); await mkdir(dirname(path), { recursive: true }); await writeFile(path, data); await writeFile(`${path}.meta`, JSON.stringify({ mimeType })); }, /** * @param {string} did * @param {string} cid */ async delete(did, cid) { const path = blobPath(did, cid); try { await unlink(path); await unlink(`${path}.meta`); } catch (e) { if (/** @type {NodeJS.ErrnoException} */ (e).code !== 'ENOENT') throw e; } }, /** * @param {string} did * @param {string} cid * @param {{start: number, end: number}} [range] - Inclusive byte positions */ async getStream(did, cid, range) { const path = blobPath(did, cid); /** @type {{mimeType: string}} */ let meta; try { meta = JSON.parse(await readFile(`${path}.meta`, 'utf8')); } catch (e) { if (/** @type {NodeJS.ErrnoException} */ (e).code === 'ENOENT') { return null; } throw e; } const stream = createReadStream( path, range ? { start: range.start, end: range.end } : undefined, ); return { body: /** @type {ReadableStream} */ ( Readable.toWeb(stream) ), mimeType: meta.mimeType, }; }, /** * Bytes stage in a temporary file and move under the CID at commit. The * `.tmp` directory sits beside the shard directories; a shard name is two * CID characters, so the names never collide. * @param {string} did */ async createWriter(did) { const dir = join(baseDir, did, '.tmp'); await mkdir(dir, { recursive: true }); const tempPath = join(dir, randomUUID()); const handle = await open(tempPath, 'w'); return { /** @param {Uint8Array} chunk */ async write(chunk) { await handle.write(chunk); }, /** * @param {string} cid * @param {string} mimeType */ async commit(cid, mimeType) { await handle.close(); const path = blobPath(did, cid); await mkdir(dirname(path), { recursive: true }); await rename(tempPath, path); await writeFile(`${path}.meta`, JSON.stringify({ mimeType })); }, async abort() { await handle.close(); try { await unlink(tempPath); } catch (e) { if (/** @type {NodeJS.ErrnoException} */ (e).code !== 'ENOENT') { throw e; } } }, }; }, }; } /** * Backups in a directory — ideally one on a different disk than the PDS * itself. Keys map straight to paths under `baseDir`; content types have * nowhere to live on a filesystem and are dropped. * @param {string} baseDir * @returns {BackupTargetPort} */ export function createFsBackupTarget(baseDir) { /** * @param {string} key * @returns {string} */ function keyPath(key) { return join(baseDir, ...key.split('/')); } return { /** * @param {string} key * @param {Uint8Array} data */ async put(key, data) { const path = keyPath(key); await mkdir(dirname(path), { recursive: true }); await writeFile(path, data); }, /** @param {string} key */ async get(key) { try { return new Uint8Array(await readFile(keyPath(key))); } catch (e) { if (/** @type {NodeJS.ErrnoException} */ (e).code === 'ENOENT') return null; throw e; } }, /** @param {string} prefix */ async list(prefix) { /** @type {import('node:fs').Dirent[]} */ let entries; try { entries = await readdir(baseDir, { recursive: true, withFileTypes: true, }); } catch (e) { if (/** @type {NodeJS.ErrnoException} */ (e).code === 'ENOENT') return []; throw e; } return entries .filter((entry) => entry.isFile()) .map((entry) => relative(baseDir, join(entry.parentPath, entry.name)) .split(sep) .join('/'), ) .filter((key) => key.startsWith(prefix)); }, /** @param {string} key */ async delete(key) { try { await unlink(keyPath(key)); } catch (e) { if (/** @type {NodeJS.ErrnoException} */ (e).code !== 'ENOENT') throw e; } }, }; }