diff --git a/core/src/configurator.ts b/core/src/configurator.ts
new file mode 100644
index 0000000..b02b959
--- /dev/null
+++ b/core/src/configurator.ts
@@ -0,0 +1,424 @@
+// Strongly-typed, remotely-updatable application configuration.
+//
+// A `Configurator` holds a set of configuration *options*, each one defined in
+// code with a `Schema` (so it is strongly typed and can carry complex values)
+// and a default value. Options can be read **synchronously** because their
+// current values are kept in an in-memory cache; the cache is kept fresh by
+// polling a durable `ConfigStore` in the background.
+//
+// Design goals (and how they are met):
+//
+// - Strongly typed, complex values → every option carries a `Schema`;
+// `define` returns a `ConfigOption` and `get()` returns `A`.
+// - Hundreds of options → all options share one store; a single
+// `load()` reconciles the whole cache.
+// - Introduced in code only → calling `define(...)` is enough. The
+// default lives in code; the store only ever holds explicit *overrides*, so
+// nothing has to be created in the database first. `reset()` deletes the
+// override and reverts to the code default.
+// - Durable → overrides live in the `ConfigStore`
+// (e.g. a database table) and survive restarts.
+// - Auto-updated, bounded staleness → a background poll (default every 10s)
+// reconciles the cache, so a change is visible in every running instance
+// within `pollIntervalMs`.
+// - Synchronous access → `get()` reads an in-memory cache. The
+// store is loaded before any option is defined, so `define()` seeds each
+// option's current value synchronously — its stored override if present,
+// otherwise the default — with no transient default value.
+// - Change callbacks → `onChange` fires with `(next, previous)`
+// whenever an option's effective value changes (remote change, `set`, or
+// `reset`); it returns an unsubscribe function.
+//
+// `@ambarltd/core` has no database dependency, so durability is abstracted
+// behind `ConfigStore`. A consuming project supplies the adapter. A minimal
+// Postgres adapter looks like this:
+//
+// const store: ConfigStore = {
+// async load() {
+// const { rows } = await pg.query(`SELECT key, value FROM configuration`);
+// return rows.map(r => ({ key: r.key, value: r.value as Json }));
+// },
+// async save(key, value) {
+// await pg.query(
+// `INSERT INTO configuration (key, value, updated_at)
+// VALUES ($1, $2::jsonb, now())
+// ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = now()`,
+// [key, JSON.stringify(value)],
+// );
+// },
+// async remove(key) {
+// await pg.query(`DELETE FROM configuration WHERE key = $1`, [key]);
+// },
+// };
+//
+// Usage — `start` the configurator (which loads the store), then `define`
+// options on the started instance:
+//
+// const configurator = await Configurator.start({ store });
+//
+// const maxUploadBytes = configurator.define({
+// key: "upload.maxBytes",
+// schema: s.number,
+// default: 10 * 1024 * 1024,
+// });
+// const featureFlags = configurator.define({
+// key: "features",
+// schema: s.object({ newCheckout: s.boolean, betaSearch: s.boolean }),
+// default: { newCheckout: false, betaSearch: false },
+// });
+//
+// maxUploadBytes.get(); // synchronous read, typed `number`
+// featureFlags.onChange((next, prev) => rebuildSearchIndex(next.betaSearch));
+// await featureFlags.set({ newCheckout: true, betaSearch: true }); // persist + propagate
+//
+// Or scope the lifetime with `Configurator.with`, which stops polling when the
+// body settles:
+//
+// await Configurator.with({ store }, configurator =>
+// Future.attemptP(async () => {
+// const flags = configurator.define({ ... });
+// // ... use flags ...
+// }),
+// );
+
+export {
+ Configurator,
+ InMemoryConfigStore,
+ type ConfigOption,
+ type ConfigOptionDef,
+ type ConfigStore,
+ type StoredConfig,
+ type ConfigChangeListener,
+ type ConfiguratorOptions,
+};
+
+import * as s from "./json/schema";
+import { Json } from "./json/types";
+import { Future } from "./future";
+import { fromOptional } from "./maybe";
+
+/** Called with the new and previous effective values when an option changes. */
+type ConfigChangeListener = (next: A, previous: A) => void;
+
+/** A single persisted override: the option's key and its JSON-encoded value. */
+type StoredConfig = { key: string; value: Json };
+
+/**
+ * Durable backing store for configuration overrides.
+ *
+ * Only options whose value has been explicitly `set` are stored — an absent key
+ * means "use the code-defined default". Implementations are expected to be
+ * shared across all running instances (e.g. a database table) so that a change
+ * made by one instance becomes visible to the others.
+ */
+interface ConfigStore {
+ /** Return every stored override. Keys not present fall back to their default. */
+ load(): Promise>;
+ /** Persist (insert or update) the override for `key`. */
+ save(key: string, value: Json): Promise;
+ /** Remove the override for `key`, reverting it to its code-defined default. */
+ remove(key: string): Promise;
+}
+
+/** Definition of a single configuration option, supplied to `Configurator.define`. */
+type ConfigOptionDef = {
+ /** Globally unique key; also the storage key in the `ConfigStore`. */
+ key: string;
+ /** Schema used to encode/decode the value to and from the store. */
+ schema: s.Schema;
+ /** Value used when no override is present in the store. */
+ default: A;
+ /** Optional human-readable description (handy for admin UIs / tooling). */
+ description?: string;
+};
+
+/** A handle to one configuration option. Reads are synchronous; writes are async. */
+interface ConfigOption {
+ readonly key: string;
+ /** The current effective value, read synchronously from the in-memory cache. */
+ get(): A;
+ /** Persist a new value to the store and update the cache (fires `onChange`). */
+ set(value: A): Promise;
+ /** Remove the override, reverting to the code default (fires `onChange`). */
+ reset(): Promise;
+ /**
+ * Register a listener invoked with `(next, previous)` whenever this option's
+ * effective value changes. Not called for the value present when the option is
+ * defined — read `get()` for that. Returns a function that unsubscribes the
+ * listener.
+ */
+ onChange(listener: ConfigChangeListener): () => void;
+}
+
+type ConfiguratorOptions = {
+ /** Durable, instance-shared backing store. */
+ store: ConfigStore;
+ /**
+ * How often, in milliseconds, to reconcile the cache against the store.
+ * Bounds cross-instance staleness. Defaults to 10_000 (10 seconds).
+ */
+ pollIntervalMs?: number;
+ /**
+ * Reports background/listener errors. Defaults to `console.error`. Background
+ * polling never throws to the caller; failures are surfaced here instead.
+ */
+ onError?: (context: string, error: Error) => void;
+};
+
+// Type-erased registry record. The public `ConfigOption` handles re-introduce
+// the precise type via casts that are sound by construction (the schema, default
+// and cached value for a given key always share the same `A`).
+type Entry = {
+ key: string;
+ schema: s.Schema;
+ default: A;
+ value: A; // current effective value — the cache (the authoritative value)
+ // The last JSON string we committed for this option — the store's form after a
+ // poll, or our encoder's form after a local write. Used purely as a byte-level
+ // cache key: if a freshly observed string equals it, nothing changed and we can
+ // skip the decode. Real-change detection compares canonically against `value`.
+ encoded: string;
+ description: string | undefined;
+ listeners: Set>;
+};
+
+/**
+ * A source of durable configuration options that get updated live.
+ */
+class Configurator {
+ private readonly registry = new Map>();
+ private timer: ReturnType | null = null;
+ private lastLoad: ReadonlyArray;
+ private polling = false; // guards against overlapping polls
+
+ private constructor(private readonly options: ConfiguratorOptions) {
+ this.lastLoad = [];
+ }
+
+ /**
+ * Run an operation with a configurator instance.
+ */
+ static with(options: ConfiguratorOptions, f: (c: Configurator) => Future): Future {
+ return Future.bracket(
+ Future.attemptP(() => Configurator.start(options)),
+ c => Future.resolve(Configurator.stop(c)),
+ f,
+ );
+ }
+
+ /**
+ * Load the store, then begin polling, returning the started configurator.
+ * Resolves once the initial load completes, so callers can `await` it before
+ * defining options or serving traffic — each subsequent `define` seeds its
+ * value from this snapshot.
+ */
+ static async start(options: ConfiguratorOptions): Promise {
+ const configurator = new Configurator(options);
+ await configurator.poll();
+
+ const pollIntervalMs = options.pollIntervalMs ?? 10_000;
+ configurator.timer = setInterval(() => {
+ void configurator.poll().catch(err => configurator.report("background poll", err));
+ }, pollIntervalMs);
+ // Don't let the poll timer keep the process alive on its own.
+ if (typeof configurator.timer.unref === "function") configurator.timer.unref();
+ return configurator;
+ }
+
+ /** Stop background polling. Cached values remain readable. */
+ static stop(configurator: Configurator): void {
+ if (configurator.timer !== null) {
+ clearInterval(configurator.timer);
+ configurator.timer = null;
+ }
+ }
+
+ /**
+ * Register a new configuration option on the started configurator. The
+ * returned handle is the only way to read or write it. The option's current
+ * value is established synchronously from the store snapshot loaded at
+ * `start()` — its stored override if present, otherwise `default` — so `get()`
+ * returns the right value immediately, with no transient default and without
+ * firing `onChange`.
+ *
+ * Throws if `key` is already defined, or if `default` cannot be encoded by
+ * `schema` (a fail-fast guard against a malformed default).
+ */
+ define(def: ConfigOptionDef): ConfigOption {
+ if (this.registry.has(def.key)) {
+ throw new Error(`Configuration option "${def.key}" is already defined`);
+ }
+
+ const entry: Entry = {
+ key: def.key,
+ schema: def.schema,
+ default: def.default,
+ value: def.default,
+ encoded: JSON.stringify(s.encode(def.schema, def.default)),
+ description: def.description,
+ listeners: new Set(),
+ };
+
+ this.registry.set(def.key, entry as Entry);
+
+ // Use value form store.
+ fromOptional(this.lastLoad.find(o => o.key === def.key))
+ .map(v => v.value)
+ .map(override => this.reconcile(entry, override));
+
+ return this.handle(entry);
+ }
+
+ /** Force an immediate reconcile against the store (fires `onChange` for any changes). */
+ async refresh(): Promise {
+ await this.poll();
+ }
+
+ // --- internals -----------------------------------------------------------
+
+ private handle(entry: Entry): ConfigOption {
+ return {
+ key: entry.key,
+ get: () => entry.value as A,
+ set: (value: A) => this.persist(entry, value),
+ reset: () => this.removeOverride(entry),
+ onChange: (listener: ConfigChangeListener) => {
+ entry.listeners.add(listener as ConfigChangeListener);
+ return () => {
+ entry.listeners.delete(listener as ConfigChangeListener);
+ };
+ },
+ };
+ }
+
+ private async persist(entry: Entry, value: A): Promise {
+ const json = s.encode(entry.schema, value); // throws on an invalid value → rejects
+ await this.options.store.save(entry.key, json);
+ this.applyValue(entry, value);
+ }
+
+ private async removeOverride(entry: Entry): Promise {
+ await this.options.store.remove(entry.key);
+ this.applyValue(entry, entry.default);
+ }
+
+ // Load all overrides and reconcile every registered option against them. A
+ // failure to load propagates to the caller (so the initial `start()` load can
+ // be awaited); a failure to reconcile a single option is isolated so one bad
+ // option can't break the rest.
+ private async poll(): Promise {
+ if (this.polling) return; // a poll is already in flight; skip this tick
+ this.polling = true;
+ try {
+ this.lastLoad = await this.options.store.load();
+ const overrides = new Map();
+ for (const { key, value } of this.lastLoad) {
+ overrides.set(key, value);
+ }
+ for (const entry of this.registry.values()) {
+ try {
+ this.reconcile(entry, overrides.get(entry.key));
+ } catch (err) {
+ this.report(`reconciling "${entry.key}"`, err);
+ }
+ }
+ } finally {
+ this.polling = false;
+ }
+ }
+
+ // Reconcile an option against its stored form.
+ private reconcile(entry: Entry, raw: Json | undefined): void {
+ const incoming = raw === undefined ? this.canonical(entry, entry.default) : JSON.stringify(raw);
+
+ // Fast path: byte-identical to the string we last committed — nothing changed,
+ // so skip the decode entirely. (After the first poll this also absorbs the
+ // store's key ordering, so it keeps hitting even when that differs from ours.)
+ if (incoming === entry.encoded) return;
+
+ // The stored bytes differ. Decode to find the effective value, then tell a
+ // real value change from a mere re-serialization (e.g. the store reorders
+ // object keys) by comparing canonically against the current value. Either
+ // way we adopt `incoming` as the new cache key so the fast path hits again.
+ const effective =
+ raw === undefined ?
+ entry.default
+ : s.decode(entry.schema, raw).either(
+ err => {
+ this.report(`decoding stored value for "${entry.key}" (using default)`, new Error(err));
+ return entry.default;
+ },
+ value => value,
+ );
+ this.commit(entry, effective, incoming);
+ }
+
+ // Set an option's value from a local write (`set`/`reset`). The cache key is
+ // our encoder's form; the next poll reconciles it to the store's form.
+ private applyValue(entry: Entry, next: A): void {
+ this.commit(entry, next, this.canonical(entry, next));
+ }
+
+ // Commit `next` (whose serialization is `serialization`) as the effective value
+ // and, when it actually changed and `notify` is set, fire listeners. Re-applying
+ // an equal value — or one that differs only in serialization — is a no-op for
+ // listeners; it still adopts `serialization` as the cache key.
+ private commit(entry: Entry, next: A, serialization: string): void {
+ if (serialization === entry.encoded) return; // identical bytes ⇒ unchanged
+
+ // Bytes differ, but the value may not — compare canonically (same key order
+ // on both sides) so a re-serialization isn't mistaken for a change.
+ const previous = entry.value;
+ const changed = this.canonical(entry, next) !== this.canonical(entry, previous);
+ entry.value = next;
+ entry.encoded = serialization;
+
+ if (!changed) return;
+ for (const listener of entry.listeners) {
+ try {
+ listener(next, previous);
+ } catch (err) {
+ this.report(`onChange listener for "${entry.key}"`, err);
+ }
+ }
+ }
+
+ // The canonical JSON string for a value, in our encoder's key order.
+ private canonical(entry: Entry, value: A): string {
+ return JSON.stringify(s.encode(entry.schema, value));
+ }
+
+ private report(context: string, error: unknown): void {
+ const err = error instanceof Error ? error : new Error(String(error));
+ if (this.options.onError) {
+ this.options.onError(context, err);
+ } else {
+ console.error(`[Configurator] ${context}:`, err);
+ }
+ }
+}
+
+/**
+ * A `ConfigStore` backed by an in-process `Map`.
+ *
+ * Useful for tests, local development, and single-process deployments. It is
+ * **not** durable across restarts and is **not** shared between instances, so it
+ * does not provide the cross-instance propagation a real deployment needs — back
+ * the `Configurator` with a shared, durable store (e.g. a database) in
+ * production.
+ */
+class InMemoryConfigStore implements ConfigStore {
+ private readonly values = new Map();
+
+ async load(): Promise> {
+ return Array.from(this.values, ([key, value]) => ({ key, value }));
+ }
+
+ async save(key: string, value: Json): Promise {
+ this.values.set(key, value);
+ }
+
+ async remove(key: string): Promise {
+ this.values.delete(key);
+ }
+}
diff --git a/core/tests/main.ts b/core/tests/main.ts
index d3c9992..1dfce99 100644
--- a/core/tests/main.ts
+++ b/core/tests/main.ts
@@ -6,6 +6,7 @@ import * as json from "@tests/suites/json";
import * as list from "@tests/suites/list";
import * as treeSet from "@tests/suites/tree-set";
import * as treeMap from "@tests/suites/tree-map";
+import * as configurator from "@tests/suites/configurator";
main();
@@ -13,5 +14,5 @@ async function main() {
console.log("Running tests");
const options = parseArgs(process.argv.slice(2));
- run(options, [time.tests, json.tests, list.tests, treeSet.tests, treeMap.tests]);
+ run(options, [time.tests, json.tests, list.tests, treeSet.tests, treeMap.tests, configurator.tests]);
}
diff --git a/core/tests/suites/configurator.ts b/core/tests/suites/configurator.ts
new file mode 100644
index 0000000..b9cb70c
--- /dev/null
+++ b/core/tests/suites/configurator.ts
@@ -0,0 +1,220 @@
+import { test, group, expect } from "test";
+import * as s from "json/schema";
+import { Json } from "json/types";
+import { Future } from "future";
+import { Configurator, InMemoryConfigStore, type ConfigStore, type ConfiguratorOptions } from "configurator";
+
+// Run a test body with a started Configurator, stopping it afterwards. Bridges
+// `Configurator.with` (a Future) into the harness's async tests.
+const withConfig = (options: ConfiguratorOptions, body: (c: Configurator) => Promise): Promise =>
+ Configurator.with(options, c => Future.attemptP(() => body(c))).promise(e => e);
+
+const flagsSchema = s.object({ newCheckout: s.boolean, betaSearch: s.boolean });
+
+const tests = group("Configurator", [
+ test("get returns the default when no override is stored", () =>
+ withConfig({ store: new InMemoryConfigStore() }, async cfg => {
+ const opt = cfg.define({ key: "a", schema: s.number, default: 42 });
+ expect.equals(opt.get(), 42);
+ })),
+
+ test("set persists, updates the cache, and fires onChange with (next, previous)", () =>
+ withConfig({ store: new InMemoryConfigStore() }, async cfg => {
+ const opt = cfg.define({ key: "a", schema: s.number, default: 1 });
+
+ const events: Array<[number, number]> = [];
+ opt.onChange((next, previous) => events.push([next, previous]));
+
+ await opt.set(5);
+ expect.equals(opt.get(), 5);
+ expect.json_equals(events, [[5, 1]]);
+ })),
+
+ test("reset removes the override and reverts to the default", () =>
+ withConfig({ store: new InMemoryConfigStore() }, async cfg => {
+ const opt = cfg.define({ key: "a", schema: s.number, default: 1 });
+ await opt.set(9);
+
+ let observed = -1;
+ opt.onChange(next => {
+ observed = next;
+ });
+
+ await opt.reset();
+ expect.equals(opt.get(), 1);
+ expect.equals(observed, 1);
+ })),
+
+ test("an option's initial value comes from the store snapshot, synchronously and silently", async () => {
+ const store = new InMemoryConfigStore();
+
+ // First run: persist an override, then shut down.
+ await withConfig({ store }, async cfg => {
+ await cfg.define({ key: "a", schema: s.number, default: 1 }).set(7);
+ });
+
+ // Second run (a "restart"): the override is loaded before any define, so the
+ // option is defined straight onto its stored value — no transient default,
+ // and no onChange for the value that was already there.
+ await withConfig({ store }, async cfg => {
+ const opt = cfg.define({ key: "a", schema: s.number, default: 1 });
+ let fired = false;
+ opt.onChange(() => {
+ fired = true;
+ });
+
+ expect.equals(opt.get(), 7); // hydrated synchronously from the snapshot
+ await cfg.refresh();
+ expect.equals(fired, false); // the pre-existing value is not a change
+ });
+ }),
+
+ test("cross-instance propagation: a poll picks up another instance's change and fires onChange", async () => {
+ const store = new InMemoryConfigStore();
+ await withConfig({ store }, async writer =>
+ withConfig({ store }, async reader => {
+ const w = writer.define({ key: "a", schema: s.number, default: 0 });
+ const r = reader.define({ key: "a", schema: s.number, default: 0 });
+
+ let observed = -1;
+ r.onChange(next => {
+ observed = next;
+ });
+
+ await w.set(123);
+ expect.equals(r.get(), 0); // the reader hasn't reconciled yet
+ await reader.refresh(); // stand in for the background poll tick
+ expect.equals(r.get(), 123);
+ expect.equals(observed, 123);
+ }),
+ );
+ }),
+
+ test("a stored value that fails to decode falls back to the default and is reported", async () => {
+ const store = new InMemoryConfigStore();
+ await store.save("a", "not a number"); // invalid for s.number
+
+ const contexts: string[] = [];
+ await withConfig({ store, onError: context => contexts.push(context) }, async cfg => {
+ const opt = cfg.define({ key: "a", schema: s.number, default: 3 });
+ expect.equals(opt.get(), 3);
+ expect.equals(contexts.length, 1);
+ });
+ }),
+
+ test("an unchanged undecodable value is reported once, not on every poll", async () => {
+ const store = new InMemoryConfigStore();
+ await store.save("a", "not a number");
+
+ const contexts: string[] = [];
+ await withConfig({ store, onError: context => contexts.push(context) }, async cfg => {
+ const opt = cfg.define({ key: "a", schema: s.number, default: 3 }); // reported once, at define
+
+ await cfg.refresh();
+ await cfg.refresh(); // stored bytes unchanged → fast path skips the decode
+
+ expect.equals(opt.get(), 3);
+ expect.equals(contexts.length, 1);
+ });
+ }),
+
+ test("defining the same key twice throws", () =>
+ withConfig({ store: new InMemoryConfigStore() }, async cfg => {
+ cfg.define({ key: "a", schema: s.number, default: 1 });
+ expect.throws(
+ () => {
+ cfg.define({ key: "a", schema: s.string, default: "x" });
+ },
+ err => expect.contains("already defined", err.message),
+ );
+ })),
+
+ test("complex object values round-trip through set and onChange", () =>
+ withConfig({ store: new InMemoryConfigStore() }, async cfg => {
+ const flags = cfg.define({
+ key: "features",
+ schema: flagsSchema,
+ default: { newCheckout: false, betaSearch: false },
+ });
+
+ let latest = flags.get();
+ flags.onChange(next => {
+ latest = next;
+ });
+
+ await flags.set({ newCheckout: true, betaSearch: false });
+ expect.json_equals(flags.get(), { newCheckout: true, betaSearch: false });
+ expect.json_equals(latest, { newCheckout: true, betaSearch: false });
+ })),
+
+ test("a key-reordered re-serialization from the store is not treated as a change", async () => {
+ // A store whose stored value can be swapped out, to mimic a backend (e.g.
+ // Postgres jsonb) handing object keys back in a different order than ours.
+ let stored: Json | undefined = undefined;
+ const store: ConfigStore = {
+ load: async () => (stored === undefined ? [] : [{ key: "features", value: stored }]),
+ save: async (_key, value) => {
+ stored = value;
+ },
+ remove: async () => {
+ stored = undefined;
+ },
+ };
+
+ await withConfig({ store }, async cfg => {
+ const flags = cfg.define({
+ key: "features",
+ schema: flagsSchema,
+ default: { newCheckout: false, betaSearch: false },
+ });
+
+ let changes = 0;
+ flags.onChange(() => {
+ changes++;
+ });
+
+ await flags.set({ newCheckout: true, betaSearch: false });
+ expect.equals(changes, 1); // a real change
+
+ // Same value, keys reversed relative to the schema's order.
+ stored = { betaSearch: false, newCheckout: true };
+ await cfg.refresh();
+ await cfg.refresh();
+
+ expect.json_equals(flags.get(), { newCheckout: true, betaSearch: false });
+ expect.equals(changes, 1); // re-serialization only — no further onChange
+ });
+ }),
+
+ test("unsubscribing stops further notifications", () =>
+ withConfig({ store: new InMemoryConfigStore() }, async cfg => {
+ const opt = cfg.define({ key: "a", schema: s.number, default: 0 });
+
+ let count = 0;
+ const off = opt.onChange(() => {
+ count++;
+ });
+
+ await opt.set(1);
+ off();
+ await opt.set(2);
+ expect.equals(count, 1);
+ })),
+
+ test("re-applying an equal value does not fire onChange", () =>
+ withConfig({ store: new InMemoryConfigStore() }, async cfg => {
+ const opt = cfg.define({ key: "a", schema: s.number, default: 0 });
+
+ let count = 0;
+ opt.onChange(() => {
+ count++;
+ });
+
+ await opt.set(5);
+ await opt.set(5); // identical value
+ await cfg.refresh(); // a poll observes the same stored value
+ expect.equals(count, 1);
+ })),
+]);
+
+export { tests };