mesh-sdk/src/state/index.ts

// State — the current value of something a module keeps on the bus, which every machine sees,
// including one that joins later (novox/hq ADR 0201). Not history: that is an event. Never a secret,
// sealed or not: a bucket is kept, and a machine joining a year later reads it whole.
//
// A module declares the state it keeps (`state` in its manifest) and the state of others it reads
// (`reads`, as `<module>.<name>`), and names each here the way it named it there. Every instance of
// the owner may write; a reader reads and watches. The runtime that launched the bundle does the work
// and refuses, with the reason, what the module did not declare.
//
// A module also says on the bus, from its code, which states it keeps and which others may read
// (`keeps`, novox/hq ADR 0495 clause 2): a reader is granted what the owner announced, never what a list
// or the owner's manifest says (clause 3).

import { emit } from "../events/index.js";

/** One key's current value. */
export interface StateEntry<T = unknown> {
  readonly key: string;
  readonly value: T;
  /** Increases with every write to the bucket; a later value has a larger one. */
  readonly revision: number;
}

/** One change a watch hands over. */
export interface StateChange<T = unknown> {
  /** The state, as the module named it. */
  readonly state: string;
  readonly key: string;
  readonly op: "put" | "delete";
  /** The new value; absent for a delete. */
  readonly value?: T;
  readonly revision: number;
  /** True for a value that was there when the watch began, false for a change since. */
  readonly current: boolean;
}

/** One state a module keeps or reads. */
export interface State<T = unknown> {
  /** A key's current value, or null when it has none. */
  get(key: string): Promise<StateEntry<T> | null>;
  /** Write a key — the owner's instances only. Resolves to the revision written. */
  put(key: string, value: T): Promise<number>;
  /** Remove a key — the owner's instances only. Removing one that is not there is not an error. */
  delete(key: string): Promise<void>;
  /** Every key with a current value, sorted. */
  keys(): Promise<string[]>;
  /**
   * Hand `handler` every current value — none that is deleted — and then every change, in order.
   * Resolves once the current values have all been handled, so code after it has the whole state.
   * `key` narrows it: a key, with `*` for one name and `**` for the rest. The watch lasts as long as
   * the process; a bundle that starts again watches again as its code runs again.
   */
  watch(handler: (change: StateChange<T>) => Promise<void> | void, opts?: { key?: string }): Promise<void>;
}

/** What the state surface needs of the channel to the runtime; the stdio session provides it. */
export interface StateChannel {
  ask(method: string, params: unknown): Promise<unknown>;
  /** Ask for a watch, with `deliver` registered for its changes before the request is sent. */
  watch(params: unknown, deliver: (change: StateChange<unknown>) => Promise<void>): Promise<void>;
}

let channel: StateChannel | undefined;

/** Bind the channel. The stdio session calls this when the runtime launched the process. */
export function useStateChannel(c: StateChannel): void {
  channel = c;
}

function boundChannel(): StateChannel {
  if (!channel) {
    throw new Error(
      "no runtime to keep state through — a module's state is reached through the runtime that launched " +
        "its bundle (novox/hq ADR 0201)",
    );
  }
  return channel;
}

/** One state a module keeps, by its local name, and whether other modules may read it. */
export interface Kept {
  readonly name: string;
  readonly readable?: boolean;
}

const stateName = /^[a-z0-9][a-z0-9-]*$/;
let keptNames: Set<string> | undefined;

/**
 * Say on the bus, as this bundle's module, every state it keeps and which of them others may read
 * (novox/hq ADR 0495 clause 2; issue 1790): the announcement `states` on the module's own event subject,
 * `{v: 1, module, node, states}` with the states sorted by name, which the controller keeps and grants
 * readers from (clause 3). Call it as the bundle is imported. From then on a state of the module's own
 * that it did not announce is refused at the call, naming it, so code and announcement cannot drift
 * (clause 5). A state no bucket could be named by, or one given twice, is refused.
 */
export async function keeps(states: readonly Kept[]): Promise<void> {
  const module = process.env.MESH_MODULE || process.env.MESH_SERVED_MODULE || "";
  if (!stateName.test(module)) {
    throw new Error(`"${module}" is not a module that keeps state: lower-case letters, digits and hyphens`);
  }
  const names = new Set<string>();
  for (const k of states) {
    if (!stateName.test(k.name)) {
      throw new Error(`${module} announces the state "${k.name}", and a state is named locally: lower-case letters, digits and hyphens`);
    }
    if (names.has(k.name)) throw new Error(`${module} announces the state "${k.name}" twice`);
    names.add(k.name);
  }
  keptNames = names;
  const sorted = [...states]
    .sort((a, b) => (a.name < b.name ? -1 : a.name > b.name ? 1 : 0))
    .map((k) => (k.readable ? { name: k.name, readable: true } : { name: k.name }));
  const node = process.env.MESH_NODE;
  await emit("states", { v: 1, module, ...(node ? { node } : {}), states: sorted });
}

/** The error of a call on a state of the module's own that `keeps` did not name, before the runtime is asked. */
function refusedUnannounced(name: string): void {
  if (!name.includes(".") && keptNames && !keptNames.has(name)) {
    throw new Error(`the state "${name}" is not among the states this module announced it keeps (keeps)`);
  }
}

/** A state this module keeps (by its local name) or reads (`<module>.<name>`). */
export function state<T = unknown>(name: string): State<T> {
  const bound = (): StateChannel => {
    refusedUnannounced(name);
    return boundChannel();
  };
  return {
    get: async (key) => (await bound().ask("mesh/state.get", { state: name, key })) as StateEntry<T> | null,
    put: async (key, value) =>
      ((await bound().ask("mesh/state.put", { state: name, key, value })) as { revision: number }).revision,
    delete: async (key) => {
      await bound().ask("mesh/state.delete", { state: name, key });
    },
    keys: async () => (await bound().ask("mesh/state.keys", { state: name })) as string[],
    watch: async (handler, opts = {}) =>
      bound().watch({ state: name, key: opts.key ?? "" }, async (change) => {
        await handler(change as StateChange<T>);
      }),
  };
}