// @pdsjs/drive - writing a stored file or folder
//
// Fulfills core's DriveWriterPort. A file's bytes arrive as a stream and leave
// as blobs without the whole file ever being held: one chunk is in memory at a
// time. Records are written straight to the ports, the same way the sites
// installer writes, rather than back over XRPC.

import {
  blobCidFromDigest,
  cborDecode,
  cborEncodeDagCbor,
  cidToString,
  createBlobCid,
  createCid,
  createTid,
} from '@pdsjs/core/repo';
import { createSha256 } from '@pdsjs/core/sha256';
import {
  DRIVE_FILE_COLLECTION,
  DRIVE_FOLDER_COLLECTION,
  isValidName,
  MAX_FOLDER_DEPTH,
  normalizeFolderKey,
} from './lexicon.js';

/** Type carried by the parts of a file that did not fit in one blob. */
const PART_MIME_TYPE = 'application/octet-stream';

/** Type a file carries when the caller names none. */
const DEFAULT_MIME_TYPE = 'application/octet-stream';

/** How many chunks one file may take, when the caller sets no other bound. */
const DEFAULT_MAX_CHUNKS = 200;

/**
 * How many bytes a client sends per request of a multi-request upload. A
 * hint, not a limit the server enforces: it sits under the smallest
 * per-request body cap a platform in front of the PDS imposes (100 MB on
 * the lowest Cloudflare plan).
 */
export const UPLOAD_PART_SIZE = 64 * 1024 * 1024;

/** How long an upload session waits between requests before it is swept. */
const UPLOAD_SESSION_TTL = 60 * 60 * 1000;

/** Upper bound on folder records read while checking a parent. */
const FOLDER_LIMIT = 2000;

/** Writes per space commit, under the cap the space writer enforces. */
const SPACE_WRITE_BATCH = 100;

/**
 * @typedef {Object} DriveWriterContext
 * @property {import('@pdsjs/core/ports').ActorStoragePort} actorStorage
 * @property {import('@pdsjs/core/ports').BlobPort} blobs
 * @property {() => Promise<string|null>} getDid
 * @property {(did: string, ops: Array<{action: string, uri: string, cid?: string}>) => Promise<Object>} createCommit
 * @property {number} chunkSize - Largest blob this server accepts, so no part exceeds it
 * @property {number} [maxFileSize] - Largest file accepted, in bytes
 * @property {import('@pdsjs/core/ports').HasherFactory} [hasher] - SHA-256 factory for the single-blob path; absent, the plain JavaScript one
 * @property {(space: string, writes: Array<{action: 'create'|'update'|'put'|'delete', collection: string, rkey: string, record?: Object}>) => Promise<unknown>} [spaceWrite] - commits into a permissioned space, injected by the platform from the spaces package. Absent, a drive can only be written in the public repo.
 * @property {import('./ports.js').DriveSpaceReader} [spaceReader] - reads space records back, so a folder can be checked before something is written into it
 */

/**
 * A space record's URI, which namespaces by the DID that wrote it because
 * every member of a space keeps their own copy. The blob link table is keyed
 * on this exact string, so it has to match what the space write route builds.
 * @param {string} space
 * @param {string} did
 * @param {string} collection
 * @param {string} rkey
 * @returns {string}
 */
function spaceRecordUri(space, did, collection, rkey) {
  return `${space}/${did}/${collection}/${rkey}`;
}

/**
 * Join a list of chunks into one array.
 * @param {Uint8Array[]} chunks
 * @param {number} size
 * @returns {Uint8Array}
 */
function join(chunks, size) {
  const joined = new Uint8Array(size);
  let offset = 0;
  for (const chunk of chunks) {
    joined.set(chunk, offset);
    offset += chunk.length;
  }
  return joined;
}

/**
 * @param {DriveWriterContext} ctx
 * @returns {import('./ports.js').DriveWriterPort}
 */
export function createDriveWriter(ctx) {
  const {
    actorStorage,
    blobs,
    getDid,
    createCommit,
    chunkSize,
    hasher = createSha256,
    spaceWrite,
    spaceReader,
  } = ctx;
  if (!actorStorage) throw new Error('createDriveWriter requires actorStorage');
  if (!blobs) throw new Error('createDriveWriter requires blobs');
  if (!getDid) throw new Error('createDriveWriter requires getDid');
  if (!createCommit) throw new Error('createDriveWriter requires createCommit');
  if (!chunkSize) throw new Error('createDriveWriter requires chunkSize');
  const maxFileSize = ctx.maxFileSize ?? chunkSize * DEFAULT_MAX_CHUNKS;

  /**
   * Every folder record, keyed by record key.
   * @param {string|null|undefined} space
   * @returns {Promise<Map<string, {name: string, parent: string, value: Record<string, unknown>}>>}
   */
  async function readFolders(space) {
    /** @type {Map<string, {name: string, parent: string, value: Record<string, unknown>}>} */
    const folders = new Map();
    /** @type {Array<{rkey: string, value: Record<string, unknown>}>} */
    const rows = [];

    if (space) {
      if (!spaceReader) return folders;
      const listing = await spaceReader.listRecords(
        space,
        DRIVE_FOLDER_COLLECTION,
        FOLDER_LIMIT,
      );
      for (const record of listing.records) {
        rows.push({
          rkey: record.rkey,
          value: /** @type {Record<string, unknown>} */ (record.value),
        });
      }
    } else {
      /** @type {string|null} */
      let cursor = null;
      while (rows.length < FOLDER_LIMIT) {
        const page = await actorStorage.listRecords(
          DRIVE_FOLDER_COLLECTION,
          cursor,
          100,
        );
        for (const record of page.records) {
          rows.push({
            rkey: record.uri.split('/').pop() || '',
            value: /** @type {Record<string, unknown>} */ (
              cborDecode(
                record.value instanceof Uint8Array
                  ? record.value
                  : new Uint8Array(record.value),
              )
            ),
          });
        }
        if (!page.cursor || page.records.length === 0) break;
        cursor = page.cursor;
      }
    }

    for (const row of rows) {
      if (typeof row.value.name !== 'string') continue;
      folders.set(row.rkey, {
        name: row.value.name,
        parent: typeof row.value.parent === 'string' ? row.value.parent : '',
        value: row.value,
      });
    }
    return folders;
  }

  /**
   * Commit one record, into the repo or into a space.
   * @param {string} did
   * @param {string|null|undefined} space
   * @param {string} collection
   * @param {string} rkey
   * @param {Record<string, unknown>} record
   * @param {string[]} [blobCids] - parts to link, for a record that holds any
   * @returns {Promise<{uri: string, cid: string}>}
   */
  async function writeRecord(did, space, collection, rkey, record, blobCids) {
    if (space && spaceWrite) {
      const uri = spaceRecordUri(space, did, collection, rkey);
      // The space commit advances the set hash. It does not index blobs:
      // only the space write route does that, so the parts are linked here
      // instead. Without the link the reap takes them, and worse, an
      // unreferenced blob is served by the public getBlob endpoint.
      const commit = /** @type {{results?: Array<{cid: string|null}>}} */ (
        await spaceWrite(space, [{ action: 'put', collection, rkey, record }])
      );
      await actorStorage.unlinkBlobsFromRecord(uri);
      for (const cid of blobCids ?? []) {
        await actorStorage.linkBlobToRecord(cid, uri, Date.now());
      }
      return { uri, cid: commit?.results?.[0]?.cid ?? '' };
    }

    const uri = `at://${did}/${collection}/${rkey}`;
    const encoded = cborEncodeDagCbor(record);
    const cid = cidToString(await createCid(encoded));
    await actorStorage.putBlock(cid, encoded);
    await actorStorage.putRecord(uri, cid, collection, rkey, encoded);
    // Linked before the commit, so a reap between the two cannot take a part
    // this record already holds. An update replaces the old links.
    await actorStorage.unlinkBlobsFromRecord(uri);
    for (const cid of blobCids ?? []) {
      await actorStorage.linkBlobToRecord(cid, uri, Date.now());
    }
    await createCommit(did, [{ action: 'create', uri, cid }]);
    return { uri, cid };
  }

  /**
   * Write several records as one commit. A cascade touches every file under a
   * folder, and one commit each would be as many commits as there are files.
   * @param {string} did
   * @param {string|null|undefined} space
   * @param {Array<{collection: string, rkey: string, record: Record<string, unknown>}>} writes
   * @returns {Promise<void>}
   */
  async function writeMany(did, space, writes) {
    if (writes.length === 0) return;

    if (space && spaceWrite) {
      // The space writer caps one commit, so a long cascade goes in batches.
      for (let at = 0; at < writes.length; at += SPACE_WRITE_BATCH) {
        await spaceWrite(
          space,
          writes.slice(at, at + SPACE_WRITE_BATCH).map((write) => ({
            action: /** @type {'put'} */ ('put'),
            collection: write.collection,
            rkey: write.rkey,
            record: write.record,
          })),
        );
      }
      return;
    }

    /** @type {Array<{action: string, uri: string, cid: string}>} */
    const ops = [];
    for (const write of writes) {
      const uri = `at://${did}/${write.collection}/${write.rkey}`;
      const encoded = cborEncodeDagCbor(write.record);
      const cid = cidToString(await createCid(encoded));
      await actorStorage.putBlock(cid, encoded);
      await actorStorage.putRecord(
        uri,
        cid,
        write.collection,
        write.rkey,
        encoded,
      );
      ops.push({ action: 'create', uri, cid });
    }
    await createCommit(did, ops);
  }

  /**
   * Remove one record, from the repo or from a space.
   * @param {string} did
   * @param {string|null|undefined} space
   * @param {string} collection
   * @param {string} rkey
   * @returns {Promise<boolean>}
   */
  async function removeRecord(did, space, collection, rkey) {
    if (space) {
      if (!spaceWrite) return false;
      try {
        await spaceWrite(space, [{ action: 'delete', collection, rkey }]);
      } catch {
        // The space write path reports a missing record by throwing; the
        // caller only needs to know it was not there.
        return false;
      }
      // Linked here on the way in, so unlinked here on the way out.
      await actorStorage.unlinkBlobsFromRecord(
        spaceRecordUri(space, did, collection, rkey),
      );
      return true;
    }

    const uri = `at://${did}/${collection}/${rkey}`;
    if (!(await actorStorage.getRecord(uri))) return false;
    await actorStorage.unlinkBlobsFromRecord(uri);
    await actorStorage.deleteRecord(uri);
    await createCommit(did, [{ action: 'delete', uri }]);
    return true;
  }

  /**
   * Every file record, as {rkey, value}.
   * @param {string|null|undefined} space
   * @returns {Promise<Array<{rkey: string, value: Record<string, unknown>}>>}
   */
  async function readFiles(space) {
    /** @type {Array<{rkey: string, value: Record<string, unknown>}>} */
    const rows = [];
    if (space) {
      if (!spaceReader) return rows;
      const listing = await spaceReader.listRecords(
        space,
        DRIVE_FILE_COLLECTION,
        FOLDER_LIMIT,
      );
      for (const record of listing.records) {
        rows.push({
          rkey: record.rkey,
          value: /** @type {Record<string, unknown>} */ (record.value),
        });
      }
      return rows;
    }
    /** @type {string|null} */
    let cursor = null;
    while (rows.length < FOLDER_LIMIT) {
      const page = await actorStorage.listRecords(
        DRIVE_FILE_COLLECTION,
        cursor,
        100,
      );
      for (const record of page.records) {
        rows.push({
          rkey: record.uri.split('/').pop() || '',
          value: /** @type {Record<string, unknown>} */ (
            cborDecode(
              record.value instanceof Uint8Array
                ? record.value
                : new Uint8Array(record.value),
            )
          ),
        });
      }
      if (!page.cursor || page.records.length === 0) break;
      cursor = page.cursor;
    }
    return rows;
  }

  /**
   * The folder and everything under it, at any depth. A folder goes to the
   * trash with what it holds, the way every drive does it, so nothing is left
   * loose at the top level with its folder gone.
   * @param {string} rkey
   * @param {Map<string, {parent: string}>} folders
   * @param {Array<{rkey: string, value: Record<string, unknown>}>} files
   * @returns {{folders: string[], files: string[]}}
   */
  function subtreeOf(rkey, folders, files) {
    const within = new Set([rkey]);
    // A parent always precedes its children in no particular order, so this
    // repeats until nothing new joins rather than assuming a walk order.
    let grew = true;
    while (grew) {
      grew = false;
      for (const [key, folder] of folders) {
        if (!within.has(key) && within.has(folder.parent)) {
          within.add(key);
          grew = true;
        }
      }
    }
    return {
      folders: [...within],
      files: files
        .filter((file) => within.has(String(file.value.folder ?? '')))
        .map((file) => file.rkey),
    };
  }

  /**
   * Reject a parent that does not exist, or a move that would put a folder
   * inside itself.
   * @param {Map<string, {parent: string}>} folders
   * @param {string} parent
   * @param {string} [moving] - the folder being moved
   * @returns {string}
   */
  function checkParent(folders, parent, moving) {
    if (!parent) return '';
    if (!folders.has(parent)) throw new Error('No such folder.');
    let at = parent;
    for (let step = 0; at && step <= MAX_FOLDER_DEPTH; step += 1) {
      if (moving && at === moving) {
        throw new Error('A folder cannot be moved inside itself.');
      }
      at = folders.get(at)?.parent ?? '';
    }
    return parent;
  }

  /**
   * Store one chunk as a blob and describe it as a record part.
   * @param {string} did
   * @param {Uint8Array} data
   * @param {string} mimeType
   */
  async function putPart(did, data, mimeType) {
    const digest = await createBlobCid(data);
    const cid = cidToString(digest);
    await actorStorage.putBlob(cid, mimeType, data.length);
    await blobs.put(did, cid, data, mimeType);
    // `{$link}` is the shape every write path in this repo stores, because
    // createRecord encodes the JSON body as it arrives. findBlobRefs finds
    // this shape and not a CID instance, so the space write path can only
    // link a part written this way.
    return {
      cid,
      part: {
        $type: 'blob',
        ref: { $link: cid },
        mimeType,
        size: data.length,
      },
    };
  }

  /**
   * @typedef {Object} StoredBytes
   * @property {Array<{$type: string, ref: {$link: string}, mimeType: string, size: number}>} parts
   * @property {string[]} partCids
   * @property {number} total
   */

  /**
   * @typedef {Object} UploadSession
   * @property {string} did
   * @property {string} folderKey
   * @property {string|null} space
   * @property {string} name
   * @property {string} type
   * @property {import('@pdsjs/core/ports').BlobWriter} writer
   * @property {import('@pdsjs/core/ports').Hasher} hash
   * @property {number} total
   * @property {number} touched
   * @property {boolean} busy
   */

  /**
   * Multi-request uploads in flight, keyed by session id. In memory: a
   * platform restart between parts fails the upload and the client starts
   * it over, which content addressing makes safe.
   * @type {Map<string, UploadSession>}
   */
  const uploads = new Map();

  /**
   * The named session, ready for its next request.
   * @param {string} id
   * @returns {UploadSession}
   */
  function takeUpload(id) {
    const session = uploads.get(id);
    if (!session) throw new Error('No such upload.');
    if (session.busy) throw new Error('That upload has a request in flight.');
    return session;
  }

  /**
   * Forget one session and abandon its staged bytes.
   * @param {string} id
   */
  async function dropUpload(id) {
    const session = uploads.get(id);
    if (!session) return;
    uploads.delete(id);
    try {
      await session.writer.abort();
    } catch {
      // staged bytes an adapter cannot reach expire on their own
    }
  }

  /** Abandon every session that has waited past the TTL. */
  async function sweepUploads() {
    const cutoff = Date.now() - UPLOAD_SESSION_TTL;
    for (const [id, session] of uploads) {
      if (session.touched < cutoff && !session.busy) await dropUpload(id);
    }
  }

  /**
   * The checks every file write shares, before any bytes move.
   * @param {string} name
   * @param {string|null|undefined} folder
   * @param {string|null|undefined} space
   * @returns {Promise<{did: string, folderKey: string}>}
   */
  async function checkFileTarget(name, folder, space) {
    const did = await getDid();
    if (!did) throw new Error('This server has no account yet.');
    if (space && !spaceWrite) {
      throw new Error('This server cannot write to a space.');
    }
    if (!isValidName(name)) throw new Error('That file name is not valid.');
    const folderKey = normalizeFolderKey(folder);
    if (folderKey === null) throw new Error('No such folder.');
    if (folderKey) {
      const folders = await readFolders(space);
      if (!folders.has(folderKey)) throw new Error('No such folder.');
    }
    return { did, folderKey };
  }

  /**
   * Write the file record over bytes already in blob storage.
   * @param {string} did
   * @param {string|null|undefined} space
   * @param {string} name
   * @param {string} folderKey
   * @param {string} type
   * @param {StoredBytes} stored
   * @returns {Promise<import('./ports.js').DriveFileRow>}
   */
  async function commitFileRecord(did, space, name, folderKey, type, stored) {
    const { parts, partCids, total } = stored;
    const rkey = createTid();
    const now = new Date().toISOString();
    const record = {
      $type: DRIVE_FILE_COLLECTION,
      name,
      ...(folderKey ? { folder: folderKey } : {}),
      parts,
      mimeType: type,
      createdAt: now,
    };
    const written = await writeRecord(
      did,
      space,
      DRIVE_FILE_COLLECTION,
      rkey,
      record,
      partCids,
    );

    return {
      rkey,
      uri: written.uri,
      cid: written.cid,
      name,
      folder: folderKey,
      mimeType: type,
      size: total,
      partCount: parts.length,
      createdAt: now,
      updatedAt: null,
    };
  }

  /**
   * Store the stream as one blob of the file's own type, so its CID is
   * something any record in the repo can reference. One chunk and one part
   * buffer are the most this holds; the blob port stages the rest.
   * @param {string} did
   * @param {ReadableStream} body
   * @param {string} type
   * @returns {Promise<StoredBytes>}
   */
  async function storeWhole(did, body, type) {
    const writer = await /** @type {NonNullable<typeof blobs.createWriter>} */ (
      blobs.createWriter
    )(did);
    const hash = hasher();
    const reader = body.getReader();
    let total = 0;
    try {
      while (true) {
        const { done, value } = await reader.read();
        if (done) break;
        const chunk =
          value instanceof Uint8Array ? value : new Uint8Array(value);
        total += chunk.length;
        if (total > maxFileSize) {
          await reader.cancel();
          throw new Error(
            `That file is larger than this server accepts (${maxFileSize} bytes).`,
          );
        }
        await hash.update(chunk);
        await writer.write(chunk);
      }
      const cid = cidToString(blobCidFromDigest(await hash.digest()));
      await actorStorage.putBlob(cid, type, total);
      await writer.commit(cid, type);
      return {
        parts: [
          { $type: 'blob', ref: { $link: cid }, mimeType: type, size: total },
        ],
        partCids: [cid],
        total,
      };
    } catch (e) {
      try {
        await writer.abort();
      } catch {
        // the original error is the one to surface
      }
      throw e;
    }
  }

  /**
   * Store the stream as chunk-sized blobs, for a port that only takes a
   * whole buffer per blob. A file past one chunk splits into opaque parts.
   * @param {string} did
   * @param {ReadableStream} body
   * @param {string} type
   * @returns {Promise<StoredBytes>}
   */
  async function storeChunked(did, body, type) {
    const reader = body.getReader();
    /** @type {Array<{$type: string, ref: {$link: string}, mimeType: string, size: number}>} */
    const parts = [];
    /** @type {string[]} */
    const partCids = [];
    /** @type {Uint8Array[]} */
    let pending = [];
    let pendingSize = 0;
    let total = 0;

    /** @param {string} partType */
    const flush = async (partType) => {
      const { cid, part } = await putPart(
        did,
        join(pending, pendingSize),
        partType,
      );
      parts.push(part);
      partCids.push(cid);
      pending = [];
      pendingSize = 0;
    };

    while (true) {
      const { done, value } = await reader.read();
      if (done) break;
      const chunk = value instanceof Uint8Array ? value : new Uint8Array(value);
      total += chunk.length;
      if (total > maxFileSize) {
        throw new Error(
          `That file is larger than this server accepts (${maxFileSize} bytes).`,
        );
      }
      pending.push(chunk);
      pendingSize += chunk.length;
      // Hold a full chunk rather than writing it: a file that ends here is
      // one part and keeps its own type, and only a second chunk makes it a
      // split file whose parts are opaque.
      while (pendingSize > chunkSize) {
        const joined = join(pending, pendingSize);
        pending = [joined.subarray(chunkSize)];
        pendingSize = pending[0].length;
        const { cid, part } = await putPart(
          did,
          joined.subarray(0, chunkSize),
          PART_MIME_TYPE,
        );
        parts.push(part);
        partCids.push(cid);
      }
    }

    // The remainder, which for a file within one chunk is the whole of it.
    // An empty file still gets a part, since the record requires one.
    await flush(parts.length === 0 ? type : PART_MIME_TYPE);
    return { parts, partCids, total };
  }

  /**
   * The item and, for a folder, everything under it. Null when there is no
   * such record.
   * @param {string} rkey
   * @param {string|null|undefined} space
   * @returns {Promise<{folders: string[], files: string[]}|null>}
   */
  async function gather(rkey, space) {
    const folders = await readFolders(space);
    const files = await readFiles(space);
    if (folders.has(rkey)) return subtreeOf(rkey, folders, files);
    if (files.some((file) => file.rkey === rkey)) {
      return { folders: [], files: [rkey] };
    }
    return null;
  }

  /**
   * Put an item in the trash, or take it out, by stamping every record it
   * covers. The parts stay linked either way: a trashed file keeps its bytes,
   * which is what makes taking it back out possible at all.
   * @param {string} rkey
   * @param {string|null|undefined} space
   * @param {string|null} at - ISO time to trash at, null to restore
   * @returns {Promise<number>} how many records the change covered
   */
  async function stamp(rkey, space, at) {
    const did = await getDid();
    if (!did) return 0;
    if (space && !spaceWrite) return 0;
    const found = await gather(rkey, space);
    if (!found) return 0;

    const folders = await readFolders(space);
    const files = await readFiles(space);
    /** @type {Array<{collection: string, rkey: string, record: Record<string, unknown>}>} */
    const writes = [];

    /** @param {string} collection @param {string} key @param {Record<string, unknown>} value */
    const change = (collection, key, value) => {
      /** @type {Record<string, unknown>} */
      const record = { ...value, $type: collection };
      if (at) record.deletedAt = at;
      else delete record.deletedAt;
      writes.push({ collection, rkey: key, record });
    };

    for (const key of found.folders) {
      const folder = folders.get(key);
      if (folder) change(DRIVE_FOLDER_COLLECTION, key, folder.value);
    }
    for (const key of found.files) {
      const file = files.find((row) => row.rkey === key);
      if (file) change(DRIVE_FILE_COLLECTION, key, file.value);
    }

    await writeMany(did, space, writes);
    return writes.length;
  }

  /**
   * @param {string} rkey
   * @param {string|null|undefined} space
   * @param {string} collection - what the caller believes it is deleting
   * @returns {Promise<boolean>}
   */
  async function trash(rkey, space, collection) {
    const folders = await readFolders(space);
    const isFolder = folders.has(rkey);
    if (collection === DRIVE_FOLDER_COLLECTION && !isFolder) return false;
    if (collection === DRIVE_FILE_COLLECTION && isFolder) return false;
    return (await stamp(rkey, space, new Date().toISOString())) > 0;
  }

  return {
    async createFolder({ name, parent, space }) {
      const did = await getDid();
      if (!did) throw new Error('This server has no account yet.');
      if (space && !spaceWrite) {
        throw new Error('This server cannot write to a space.');
      }
      if (!isValidName(name)) throw new Error('That folder name is not valid.');
      const parentKey = normalizeFolderKey(parent);
      if (parentKey === null) throw new Error('No such folder.');

      const folders = await readFolders(space);
      checkParent(folders, parentKey);
      for (const folder of folders.values()) {
        if (folder.parent === parentKey && folder.name === name) {
          throw new Error('A folder of that name is already here.');
        }
      }

      const rkey = createTid();
      const now = new Date().toISOString();
      const record = {
        $type: DRIVE_FOLDER_COLLECTION,
        name,
        ...(parentKey ? { parent: parentKey } : {}),
        createdAt: now,
      };
      const written = await writeRecord(
        did,
        space,
        DRIVE_FOLDER_COLLECTION,
        rkey,
        record,
      );
      return {
        rkey,
        uri: written.uri,
        cid: written.cid,
        name,
        parent: parentKey,
        createdAt: now,
      };
    },

    async renameFolder({ rkey, name, parent, space }) {
      const did = await getDid();
      if (!did) throw new Error('This server has no account yet.');
      if (space && !spaceWrite) {
        throw new Error('This server cannot write to a space.');
      }
      const folders = await readFolders(space);
      const folder = folders.get(rkey);
      if (!folder) throw new Error('No such folder.');
      if (name !== undefined && !isValidName(name)) {
        throw new Error('That folder name is not valid.');
      }

      let parentKey = folder.parent;
      if (parent !== undefined) {
        const requested = normalizeFolderKey(parent);
        if (requested === null) throw new Error('No such folder.');
        parentKey = checkParent(folders, requested, rkey);
      }

      const now = new Date().toISOString();
      /** @type {Record<string, unknown>} */
      const record = {
        ...folder.value,
        $type: DRIVE_FOLDER_COLLECTION,
        name: name ?? folder.name,
        updatedAt: now,
      };
      if (parentKey) record.parent = parentKey;
      else delete record.parent;

      const written = await writeRecord(
        did,
        space,
        DRIVE_FOLDER_COLLECTION,
        rkey,
        record,
      );
      return {
        rkey,
        uri: written.uri,
        cid: written.cid,
        name: /** @type {string} */ (record.name),
        parent: parentKey,
        createdAt:
          typeof folder.value.createdAt === 'string'
            ? folder.value.createdAt
            : now,
      };
    },

    async deleteFolder(rkey, space) {
      return trash(rkey, space, DRIVE_FOLDER_COLLECTION);
    },

    async restore(rkey, space) {
      return stamp(rkey, space, null);
    },

    async purge(rkey, space) {
      const did = await getDid();
      if (!did) return 0;
      const found = await gather(rkey, space);
      if (!found) return 0;
      // Unlinked here, so the orphan sweep is what frees the bytes. The
      // record goes with it: nothing is left naming a blob that is gone.
      for (const key of found.files) {
        await removeRecord(did, space, DRIVE_FILE_COLLECTION, key);
      }
      for (const key of found.folders) {
        await removeRecord(did, space, DRIVE_FOLDER_COLLECTION, key);
      }
      return found.files.length + found.folders.length;
    },

    async createFile({ name, folder, mimeType, body, space }) {
      const { did, folderKey } = await checkFileTarget(name, folder, space);
      if (!body) throw new Error('No file was sent.');

      const type = mimeType || DEFAULT_MIME_TYPE;
      const stored = blobs.createWriter
        ? await storeWhole(did, body, type)
        : await storeChunked(did, body, type);

      return commitFileRecord(did, space, name, folderKey, type, stored);
    },

    // The session surface only exists over a blob port that streams, so its
    // absence is what tells the account API to say no.
    ...(blobs.createWriter
      ? {
          uploadPartSize: UPLOAD_PART_SIZE,

          /**
           * @param {{name: string, folder: string|null, mimeType: string, space?: string|null}} file
           */
          async createUpload({ name, folder, mimeType, space }) {
            const { did, folderKey } = await checkFileTarget(
              name,
              folder,
              space,
            );
            await sweepUploads();
            const id = crypto.randomUUID();
            uploads.set(id, {
              did,
              folderKey,
              space: space || null,
              name,
              type: mimeType || DEFAULT_MIME_TYPE,
              writer:
                await /** @type {NonNullable<typeof blobs.createWriter>} */ (
                  blobs.createWriter
                )(did),
              hash: hasher(),
              total: 0,
              touched: Date.now(),
              busy: false,
            });
            return { id };
          },

          /**
           * @param {string} id
           * @param {ReadableStream<Uint8Array>} body
           */
          async appendUpload(id, body) {
            const session = takeUpload(id);
            session.busy = true;
            try {
              const reader = body.getReader();
              while (true) {
                const { done, value } = await reader.read();
                if (done) break;
                const chunk =
                  value instanceof Uint8Array ? value : new Uint8Array(value);
                session.total += chunk.length;
                if (session.total > maxFileSize) {
                  await reader.cancel();
                  throw new Error(
                    `That file is larger than this server accepts (${maxFileSize} bytes).`,
                  );
                }
                await session.hash.update(chunk);
                await session.writer.write(chunk);
              }
              session.busy = false;
              session.touched = Date.now();
              return { received: session.total };
            } catch (e) {
              // A part that fails partway has already fed the hash, so the
              // session cannot go on. The client starts the upload over.
              await dropUpload(id);
              throw e;
            }
          },

          /** @param {string} id */
          async finishUpload(id) {
            const session = takeUpload(id);
            session.busy = true;
            uploads.delete(id);
            const { did, space, name, folderKey, type, writer, hash, total } =
              session;
            try {
              const cid = cidToString(blobCidFromDigest(await hash.digest()));
              await actorStorage.putBlob(cid, type, total);
              await writer.commit(cid, type);
              const stored = {
                parts: [
                  {
                    $type: 'blob',
                    ref: { $link: cid },
                    mimeType: type,
                    size: total,
                  },
                ],
                partCids: [cid],
                total,
              };
              return await commitFileRecord(
                did,
                space,
                name,
                folderKey,
                type,
                stored,
              );
            } catch (e) {
              try {
                await writer.abort();
              } catch {
                // the original error is the one to surface
              }
              throw e;
            }
          },

          /** @param {string} id */
          async abortUpload(id) {
            await dropUpload(id);
          },
        }
      : {}),

    async deleteFile(rkey, space) {
      return trash(rkey, space, DRIVE_FILE_COLLECTION);
    },
  };
}