chadtmiller.com/pds.js

Switch repository

Loading…

// @pdsjs/spaces/writer - the single write path for permissioned repos.
//
// createRecord, putRecord, deleteRecord and applyWrites all funnel through
// applyWrites here: one commit, one rev, one LtHash advance, one atomic
// commitSpaceWrite. Nothing else mutates a permissioned repo.

import {
  cborEncodeDagCbor,
  cidToString,
  createCid,
  createTid,
  jsonToLex,
} from '@pdsjs/core/repo';
import { RepoCommit } from './commit.js';
import { formatRecordPath } from './path.js';
import { makeSpaceRow } from './space-row.js';
import { MAX_RECORD_SIZE } from './star.js';
import { parseSpaceUri } from './uri.js';

export const MAX_WRITES_PER_COMMIT = 200;

export class SpaceWriteError extends Error {
  /**
   * @param {string} message
   * @param {string} code - machine-readable error name for the XRPC response
   */
  constructor(message, code) {
    super(message);
    this.name = 'SpaceWriteError';
    this.code = code;
  }
}

export class SpaceRecordNotFoundError extends SpaceWriteError {
  /** @param {string} collection @param {string} rkey */
  constructor(collection, rkey) {
    super(`Record not found: ${collection}/${rkey}`, 'RecordNotFound');
  }
}

export class SpaceRecordAlreadyExistsError extends SpaceWriteError {
  /** @param {string} collection @param {string} rkey */
  constructor(collection, rkey) {
    super(
      `Record already exists: ${collection}/${rkey}`,
      'RecordAlreadyExists',
    );
  }
}

// Commits for one space must not interleave: applyWrites reads the current set
// hash and the touched records' CIDs, then commits a state derived from them.
// Two concurrent commits would each fold their ops into the same starting hash
// and the last write would drop the other's contribution. The port's commit is
// atomic but cannot serialise across calls, so serialise here — sound because a
// permissioned repo lives in exactly one process (a Node PDS, or one Durable
// Object).
/** @type {WeakMap<object, Map<string, Promise<any>>>} */
const spaceQueues = new WeakMap();

/**
 * @template T
 * @param {object} storage
 * @param {string} space
 * @param {() => Promise<T>} fn
 * @returns {Promise<T>}
 */
function serialize(storage, space, fn) {
  let queues = spaceQueues.get(storage);
  if (!queues) {
    queues = new Map();
    spaceQueues.set(storage, queues);
  }
  const prior = queues.get(space) ?? Promise.resolve();
  // Chain off the prior commit's settlement, not its value, so one failed
  // commit does not poison every later write to the same space.
  const next = prior.then(fn, fn);
  queues.set(
    space,
    next.then(
      () => undefined,
      () => undefined,
    ),
  );
  return next;
}

/**
 * @typedef {Object} SpaceWriteInput
 * @property {'create'|'update'|'put'|'delete'} action
 * @property {string} collection
 * @property {string} rkey
 * @property {Object} [record] - required for everything but delete
 * @property {string|null} [swapCid] - compare-and-swap: the CID the record must
 *   currently have, or null to require that it does not exist. Absent means no
 *   check. Checked inside the per-space queue, so the comparison and the commit
 *   cannot interleave with another write.
 */

/**
 * @typedef {Object} SpaceWriteResult
 * @property {'create'|'update'|'delete'} action
 * @property {string} collection
 * @property {string} rkey
 * @property {string|null} cid - null for deletes
 * @property {string|null} prev - null for creates
 */

/**
 * Apply a batch of writes to a permissioned repo as one commit.
 *
 * Writes resolve in order against the state the batch has built up, not just
 * what was stored on entry, so a batch can create then update a record and a
 * repeated create is still caught rather than double-counted in the set hash.
 *
 * @param {import('@pdsjs/core/ports').SpaceStoragePort} storage
 * @param {Object} opts
 * @param {string} opts.space - space AT-URI
 * @param {SpaceWriteInput[]} opts.writes
 * @param {string|null} [opts.owner] - the hosted account. A write into a space
 *   it is the authority of needs that space to exist and not be deleted.
 * @param {string} [opts.now] - ISO timestamp, for deterministic tests
 * @returns {Promise<{rev: string, setHash: Uint8Array, results: SpaceWriteResult[]}>}
 */
export async function applyWrites(storage, { space, writes, owner, now }) {
  if (writes.length > MAX_WRITES_PER_COMMIT) {
    throw new SpaceWriteError(
      `Too many writes. Max: ${MAX_WRITES_PER_COMMIT}`,
      'InvalidRequest',
    );
  }

  return serialize(storage, space, async () => {
    const timestamp = now ?? new Date().toISOString();

    const existing = await storage.getSpace(space);
    // A write must neither conjure an ungoverned space nor revive a deleted
    // one, so the authority's own spaces must already exist.
    if (owner && parseSpaceUri(space).spaceDid === owner) {
      if (!existing?.isOwner || existing.deletedAt) {
        throw new SpaceWriteError('Space not found', 'SpaceNotFound');
      }
    }
    // A writer's PDS is never told about membership, which is the authority's
    // concern, so it materialises its own repo on first write.
    if (existing === null) {
      await storage.putSpace(
        makeSpaceRow(space, { isOwner: false, createdAt: timestamp }),
      );
    }

    const state = await storage.getSpaceRepo(space);
    const repo = RepoCommit.fromState(state?.setHash ?? null);
    const rev = createTid();

    /** @type {Map<string, string|null>} live cid per touched path; null = deleted */
    const staged = new Map();
    /** @type {SpaceWriteResult[]} */
    const results = [];
    /** @type {import('@pdsjs/core/ports').SpaceWrite[]} */
    const portWrites = [];

    for (const write of writes) {
      const { collection, rkey } = write;
      const key = formatRecordPath(collection, rkey);
      const prev = staged.has(key)
        ? /** @type {string|null} */ (staged.get(key))
        : await storage.getSpaceRecordCid(space, collection, rkey);

      if (write.swapCid !== undefined && write.swapCid !== prev) {
        throw new SpaceWriteError(
          `Record ${collection}/${rkey} is at ${prev ?? 'no version'}, expected ${write.swapCid ?? 'no record'}`,
          'InvalidSwap',
        );
      }
      if (write.action === 'create' && prev) {
        throw new SpaceRecordAlreadyExistsError(collection, rkey);
      }
      if ((write.action === 'update' || write.action === 'delete') && !prev) {
        throw new SpaceRecordNotFoundError(collection, rkey);
      }

      if (write.action === 'delete') {
        repo.applyOp({ collection, rkey, cid: null, prev });
        staged.set(key, null);
        results.push({ action: 'delete', collection, rkey, cid: null, prev });
        portWrites.push({
          action: 'delete',
          collection,
          rkey,
          cid: null,
          value: null,
          prev,
        });
        continue;
      }

      const value = cborEncodeDagCbor(jsonToLex(write.record));
      if (value.length > MAX_RECORD_SIZE) {
        throw new SpaceWriteError(
          `Record too large: ${value.length} bytes, max ${MAX_RECORD_SIZE}`,
          'InvalidRequest',
        );
      }
      const cid = cidToString(await createCid(value));
      repo.applyOp({ collection, rkey, cid, prev });
      staged.set(key, cid);

      const action = prev ? 'update' : 'create';
      results.push({ action, collection, rkey, cid, prev });
      portWrites.push({ action, collection, rkey, cid, value, prev });
    }

    const setHash = repo.state();
    await storage.commitSpaceWrite(space, {
      writes: portWrites,
      setHash,
      rev,
      indexedAt: timestamp,
    });

    return { rev, setHash, results };
  });
}