harbor-net/src/server/services/stateStore.js

import crypto from 'node:crypto';
import fs from 'node:fs';
import path from 'node:path';
import { normalizeStoredState } from '../../shared/contracts/state.js';
import { INITIAL_ROUTE_RULES } from '../../shared/routingRules.js';

export const STATE_SCHEMA_VERSION = 4;

const clone = (value) => structuredClone(value);
const stamp = (value) => value.toISOString().replace(/[:.]/g, '-');

function syncDirectory(directory) {
  let descriptor;
  try {
    descriptor = fs.openSync(directory, 'r');
    fs.fsyncSync(descriptor);
  } catch (error) {
    if (!['EINVAL', 'ENOTSUP', 'EPERM'].includes(error.code)) throw error;
  } finally {
    if (descriptor !== undefined) fs.closeSync(descriptor);
  }
}

export function atomicWriteFile(filePath, contents, { beforeRename, mode } = {}) {
  const directory = path.dirname(filePath);
  fs.mkdirSync(directory, { recursive: true });
  const temporaryPath = path.join(
    directory,
    `.${path.basename(filePath)}.${process.pid}.${crypto.randomUUID()}.tmp`,
  );
  const fileMode = mode ?? (fs.existsSync(filePath) ? fs.statSync(filePath).mode & 0o777 : 0o666);
  let descriptor;

  try {
    descriptor = fs.openSync(temporaryPath, 'wx', fileMode);
    fs.writeFileSync(descriptor, contents, 'utf8');
    fs.fsyncSync(descriptor);
    fs.closeSync(descriptor);
    descriptor = undefined;
    beforeRename?.(temporaryPath, filePath);
    fs.renameSync(temporaryPath, filePath);
    syncDirectory(directory);
  } finally {
    if (descriptor !== undefined) fs.closeSync(descriptor);
    fs.rmSync(temporaryPath, { force: true });
  }
}

export function atomicWriteJson(filePath, value, options) {
  atomicWriteFile(filePath, JSON.stringify(value, null, 2), options);
}

export function migrateStoredState(value) {
  const stored = value && typeof value === 'object' && !Array.isArray(value) ? value : {};
  const version = Number.isSafeInteger(stored.schemaVersion) ? stored.schemaVersion : 0;
  if (version < 0 || version > STATE_SCHEMA_VERSION) {
    throw new Error(`Unsupported Harbor state schemaVersion: ${version}`);
  }
  const routeRules = version < 3
    ? [...INITIAL_ROUTE_RULES, ...(Array.isArray(stored.routeRules) ? stored.routeRules : [])]
    : stored.routeRules;
  return {
    ...normalizeStoredState({ ...stored, routeRules }),
    schemaVersion: STATE_SCHEMA_VERSION,
  };
}

export function createJsonStore({
  filePath,
  defaultValue,
  migrate = (value) => value,
  initializeMissing = false,
  backupWhen = () => false,
  now = () => new Date(),
} = {}) {
  let recovery = null;
  let migration = null;

  function write(value, options) {
    const migrated = migrate(clone(value));
    atomicWriteJson(filePath, migrated, options);
    return clone(migrated);
  }

  function read() {
    if (!fs.existsSync(filePath)) {
      const initial = migrate(clone(defaultValue));
      return initializeMissing ? write(initial) : clone(initial);
    }

    const raw = fs.readFileSync(filePath, 'utf8');
    let parsed;
    try {
      parsed = JSON.parse(raw);
    } catch (cause) {
      const backupPath = `${filePath}.corrupt-${stamp(now())}`;
      fs.renameSync(filePath, backupPath);
      try {
        const recovered = write(defaultValue);
        recovery = { kind: 'corrupt-json', backupPath, recoveredAt: now().toISOString() };
        return recovered;
      } catch (error) {
        fs.renameSync(backupPath, filePath);
        throw new AggregateError([cause, error], `Failed to recover corrupt JSON: ${filePath}`);
      }
    }

    const migrated = migrate(clone(parsed));
    if (JSON.stringify(migrated) !== JSON.stringify(parsed)) {
      if (backupWhen(parsed, migrated)) {
        const fromVersion = Number.isSafeInteger(parsed?.schemaVersion) ? parsed.schemaVersion : 0;
        const backupPath = `${filePath}.backup-v${fromVersion}-${stamp(now())}`;
        atomicWriteFile(backupPath, raw, { mode: fs.statSync(filePath).mode & 0o777 });
        migration = {
          fromVersion,
          toVersion: migrated.schemaVersion,
          backupPath,
          migratedAt: now().toISOString(),
        };
      }
      atomicWriteJson(filePath, migrated);
    }
    return clone(migrated);
  }

  function update(mutator) {
    // ponytail: sync mutators serialize in Node's event loop; add a queue only if updates must await I/O.
    const next = mutator(read());
    if (next && typeof next.then === 'function') {
      throw new TypeError('State store mutator must be synchronous');
    }
    return write(next);
  }

  return {
    read,
    write,
    update,
    remove: () => fs.rmSync(filePath, { force: true }),
    get recovery() { return recovery; },
    get migration() { return migration; },
  };
}

export function createStateStore(filePath, options = {}) {
  return createJsonStore({
    filePath,
    defaultValue: {},
    migrate: migrateStoredState,
    initializeMissing: true,
    backupWhen: (before, after) => before?.schemaVersion !== after.schemaVersion,
    ...options,
  });
}