chadtmiller.com/pds.js

Switch repository

Loading…

// @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<Uint8Array>} */ (
          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;
      }
    },
  };
}