/** * 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} 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} cursors - PDS base URL to sequence * @property {Record>} refs - "did/name" to ref map * @property {Record>} 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} 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} */ refs(key) { return state.refs[key] ?? {}; }, /** * @param {string} key * @param {Record} 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(); }, }; }