chadtmiller.com/pds.js

Switch repository

Loading…

/**
 * What the daemon must remember across restarts: the firehose sequence it
 * reached on each PDS, the object id it last saw on every ref, and the
 * version of every check request it has acted on.
 *
 * All three are needed. The cursor decides which events replay, and the other
 * two decide which of those events describe work. Without the ref map, a
 * replay would re-run every ref in the record rather than the one that moved.
 * Without the request map, it would re-run every request ever made.
 */

import { mkdirSync, readFileSync, writeFileSync } from 'node:fs';
import { homedir } from 'node:os';
import { dirname, join } from 'node:path';

/**
 * @param {Record<string, string|undefined>} env
 * @returns {string}
 */
export function statePath(env) {
  if (env.PDSJS_CI_STATE) return env.PDSJS_CI_STATE;
  const base = env.XDG_STATE_HOME || join(homedir(), '.local', 'state');
  return join(base, 'pdsjs-git-ci', 'state.json');
}

/**
 * @typedef {Object} DaemonState
 * @property {Record<string, number>} cursors - PDS base URL to sequence
 * @property {Record<string, Record<string, string>>} refs - "did/name" to ref map
 * @property {Record<string, Record<string, string>>} requests - "did/name" to request rkey to record cid
 */

/**
 * @param {string} path
 * @returns {DaemonState}
 */
function readState(path) {
  try {
    const parsed = JSON.parse(readFileSync(path, 'utf8'));
    return {
      cursors: parsed.cursors ?? {},
      refs: parsed.refs ?? {},
      requests: parsed.requests ?? {},
    };
  } catch {
    return { cursors: {}, refs: {}, requests: {} };
  }
}

/** @typedef {ReturnType<typeof createStateStore>} StateStore */

/**
 * @param {string} path
 * @param {(message: string) => void} [onWarn]
 */
export function createStateStore(path, onWarn = () => {}) {
  const state = readState(path);
  let unwritable = false;

  // A state file that cannot be written costs durability, not the run. The
  // daemon keeps the cursor and the ref map in memory and carries on, so an
  // unwritable path degrades to "replays from the live head after a restart"
  // rather than stopping the events that reach the runner.
  function flush() {
    if (unwritable) return;
    try {
      mkdirSync(dirname(path), { recursive: true });
      writeFileSync(path, `${JSON.stringify(state, null, 2)}\n`, {
        mode: 0o600,
      });
    } catch (err) {
      unwritable = true;
      onWarn(
        `state file ${path} is not writable (${err instanceof Error ? err.message : err}); continuing without saving progress`,
      );
    }
  }

  return {
    /**
     * @param {string} service
     * @returns {number|undefined}
     */
    cursor(service) {
      return state.cursors[service];
    },

    /**
     * @param {string} service
     * @param {number} seq
     */
    setCursor(service, seq) {
      state.cursors[service] = seq;
      flush();
    },

    /**
     * @param {string} key
     * @returns {Record<string, string>}
     */
    refs(key) {
      return state.refs[key] ?? {};
    },

    /**
     * @param {string} key
     * @param {Record<string, string>} refs
     */
    setRefs(key, refs) {
      state.refs[key] = refs;
      flush();
    },

    /**
     * The record version of each check request already acted on, by request
     * record key. A request keeps its key and changes its version every time
     * it is made, so the cid tells a fresh request from a replayed one.
     * @param {string} key
     * @param {string} rkey
     * @returns {string|undefined}
     */
    request(key, rkey) {
      return state.requests[key]?.[rkey];
    },

    /**
     * @param {string} key
     * @param {string} rkey
     * @param {string} cid
     */
    setRequest(key, rkey, cid) {
      state.requests[key] = { ...state.requests[key], [rkey]: cid };
      flush();
    },
  };
}