chadtmiller.com/pds.js

Switch repository

Loading…

// @pdsjs/sites - materialized queries over the repo
//
// A dev.pdsjs.query.def record names a live view over one collection;
// GET /xrpc/dev.pdsjs.query.run serves its rows as JSON in one request. The
// engine keeps a decoded copy of each queried collection in memory: loaded
// from storage on first use, maintained from the core commit tap, rebuilt
// by replay after a restart. Writes that bypass createCommit (a CAR import)
// are picked up on the restart that follows them.
//
// Queries read the public repo only. Space records live in their own
// storage and never appear here.

import { cborDecode, toDisplayValue } from '@pdsjs/core/repo';
import { QUERY_COLLECTION } from './lexicons.js';

const DEFAULT_LIMIT = 100;
const MAX_LIMIT = 2000;
const LOAD_PAGE = 100;

/**
 * Read a dot-path off a decoded record.
 * @param {unknown} value
 * @param {string} path
 * @returns {unknown}
 */
export function fieldAt(value, path) {
  let current = value;
  for (const segment of path.split('.')) {
    if (current === null || typeof current !== 'object') return undefined;
    current = /** @type {Record<string, unknown>} */ (current)[segment];
  }
  return current;
}

/**
 * @typedef {Object} QueryEngineContext
 * @property {import('@pdsjs/core/ports').ActorStoragePort} actorStorage
 * @property {() => Promise<string|null>} getDid
 */

/**
 * @typedef {Object} CachedRecord
 * @property {string} uri
 * @property {string} rkey
 * @property {string} cid
 * @property {unknown} value - decoded record, JSON-shaped
 */

/**
 * Build the query engine: a route table fragment for core's queryRoutes and
 * an onCommit subscriber that keeps the caches current.
 *
 * @param {QueryEngineContext} ctx
 * @returns {{
 *   routes: Record<string, {method?: string, handler: (request: Request, url: URL) => Promise<Response>}>,
 *   onCommit: import('@pdsjs/core/ports').CommitTapPort,
 * }}
 */
export function createQueryEngine(ctx) {
  const { actorStorage, getDid } = ctx;
  if (!actorStorage) throw new Error('createQueryEngine requires actorStorage');
  if (!getDid) throw new Error('createQueryEngine requires getDid');

  /** @type {Map<string, Map<string, CachedRecord>>} */
  const collections = new Map();
  /** Loads in flight, so concurrent first requests replay a collection once. */
  /** @type {Map<string, Promise<Map<string, CachedRecord>>>} */
  const loading = new Map();

  /**
   * @param {string} uri
   * @param {string} cid
   * @param {Uint8Array|ArrayBuffer} value
   * @returns {CachedRecord}
   */
  function decodeRow(uri, cid, value) {
    const bytes = value instanceof Uint8Array ? value : new Uint8Array(value);
    return {
      uri,
      rkey: uri.split('/').pop() || '',
      cid,
      value: toDisplayValue(cborDecode(bytes)),
    };
  }

  /**
   * The decoded rows of a collection, replayed from storage on first use.
   * @param {string} collection
   * @returns {Promise<Map<string, CachedRecord>>}
   */
  async function loadCollection(collection) {
    const cached = collections.get(collection);
    if (cached) return cached;
    const inFlight = loading.get(collection);
    if (inFlight) return inFlight;

    const load = (async () => {
      /** @type {Map<string, CachedRecord>} */
      const rows = new Map();
      let cursor = /** @type {string|null} */ (null);
      do {
        const page = await actorStorage.listRecords(
          collection,
          cursor,
          LOAD_PAGE,
        );
        for (const record of page.records) {
          rows.set(record.uri, decodeRow(record.uri, record.cid, record.value));
        }
        cursor = page.records.length ? page.cursor : null;
      } while (cursor);
      collections.set(collection, rows);
      return rows;
    })();
    loading.set(collection, load);
    try {
      return await load;
    } finally {
      loading.delete(collection);
    }
  }

  /** @type {import('@pdsjs/core/ports').CommitTapPort} */
  async function onCommit(evt) {
    for (const op of evt.ops) {
      const [collection] = op.uri.replace(`at://${evt.did}/`, '').split('/');
      // A collection nobody has queried yet has no cache to maintain; the
      // first query replays it from storage, this write included.
      const rows = collections.get(collection);
      if (!rows) continue;
      if (op.action === 'delete') {
        rows.delete(op.uri);
      } else {
        const stored = await actorStorage.getRecord(op.uri);
        if (stored) {
          rows.set(op.uri, decodeRow(op.uri, stored.cid, stored.value));
        }
      }
    }
  }

  /**
   * @param {string} error
   * @param {string} message
   * @param {number} status
   */
  const errorResponse = (error, message, status) =>
    Response.json({ error, message }, { status });

  /**
   * GET /xrpc/dev.pdsjs.query.run?query=<name>&limit&cursor
   * @param {Request} _request
   * @param {URL} url
   * @returns {Promise<Response>}
   */
  async function handleQueryRun(_request, url) {
    const name = url.searchParams.get('query');
    if (!name) {
      return errorResponse(
        'InvalidRequest',
        'Missing required parameter: query',
        400,
      );
    }

    const did = await getDid();
    if (!did) {
      return errorResponse('QueryNotFound', 'No such query.', 404);
    }

    const defs = await loadCollection(QUERY_COLLECTION);
    const def = defs.get(`at://${did}/${QUERY_COLLECTION}/${name}`);
    if (!def) {
      return errorResponse('QueryNotFound', 'No such query.', 404);
    }
    const definition =
      /** @type {{collection?: unknown, where?: unknown, sort?: {field?: unknown, direction?: unknown}}} */ (
        def.value
      );
    if (typeof definition.collection !== 'string') {
      return errorResponse(
        'InvalidQuery',
        'The query definition names no collection.',
        400,
      );
    }

    const limit = Math.min(
      MAX_LIMIT,
      Math.max(1, Number(url.searchParams.get('limit')) || DEFAULT_LIMIT),
    );
    const offset = Math.max(0, Number(url.searchParams.get('cursor')) || 0);

    let rows = [...(await loadCollection(definition.collection)).values()];

    if (definition.where && typeof definition.where === 'object') {
      const conditions = Object.entries(definition.where);
      rows = rows.filter((row) =>
        conditions.every(
          ([path, expected]) => fieldAt(row.value, path) === expected,
        ),
      );
    }

    const sortField =
      typeof definition.sort?.field === 'string' ? definition.sort.field : null;
    const descending = definition.sort?.direction !== 'asc';
    rows.sort((a, b) => {
      if (sortField) {
        const left = fieldAt(a.value, sortField);
        const right = fieldAt(b.value, sortField);
        // Missing values sort last in either direction
        if (left === undefined && right !== undefined) return 1;
        if (right === undefined && left !== undefined) return -1;
        if (left !== right) {
          const cmp =
            /** @type {any} */ (left) < /** @type {any} */ (right) ? -1 : 1;
          return descending ? -cmp : cmp;
        }
      }
      // Record keys are TIDs in every collection queries target, so the
      // tiebreak (and the default order) is creation order, newest first.
      const cmp = a.rkey < b.rkey ? -1 : a.rkey > b.rkey ? 1 : 0;
      return sortField && !descending ? cmp : -cmp;
    });

    const page = rows.slice(offset, offset + limit);
    /** @type {{records: Array<{uri: string, cid: string, value: unknown}>, cursor?: string}} */
    const body = {
      records: page.map(({ uri, cid, value }) => ({ uri, cid, value })),
    };
    if (offset + limit < rows.length) body.cursor = String(offset + limit);
    return Response.json(body);
  }

  return {
    routes: {
      '/xrpc/dev.pdsjs.query.run': { method: 'GET', handler: handleQueryRun },
    },
    onCommit,
  };
}