23111218316

Committed 15 Mar 2026 01:21PM UTC coverage: 87.12% (+0.2%) from 86.926%

Build # 23111218316

Build Type

push

github

Committed by

web-flow

Commit Message

Use in-process event emitter instead of Redis pub/sub (#741)

Coverage Stats

1026 of 1214 branches covered (84.51%)

Branch coverage included in aggregate %.

69 of 71 new or added lines in 6 files covered. (97.18%)

10676 of 12218 relevant lines covered (87.38%)

107.98 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.22

/src/plugins/configStore.js

import assert from 'node:assert';
import fs from 'node:fs';
import EventEmitter from 'node:events';
import Ajv from 'ajv/dist/2019.js';
import formats from 'ajv-formats';
import jsonMergePatch from 'json-merge-patch';
import ValidationError from '../errors/ValidationError.js';
import { sql } from 'kysely';
import { fromJson, json, jsonb } from '../utils/sqlite.js';

/**
 * @typedef {import('type-fest').JsonObject} JsonObject
 * @typedef {import('../schema.js').UserID} UserID
 * @typedef {import('../schema.js').User} User
 */

const CONFIG_UPDATE_MESSAGE = 'configStore:update';

/**
 * Extensible configuration store.
 *
 * The config store contains named groups of settings. Each setting group is
 * stored in its own MongoDB Document. Groups have associated JSON Schemas to
 * check that the configuration is correct.
 */
class ConfigStore {
  #uw;

  #logger;

  #ajv;

  #emitter = new EventEmitter();

  #unsubscribe;

  /** @type {Map<string, import('ajv').ValidateFunction<unknown>>} */
  #validators = new Map();

  /**
   * @param {import('../Uwave.js').Boot} uw
   */
  constructor(uw) {
    this.#uw = uw;
    this.#logger = uw.logger.child({ ns: 'uwave:config' });
    this.#ajv = new Ajv({
      useDefaults: true,
      // Allow unknown keywords (`uw:xyz`)
      strict: false,
      strictTypes: true,
    });
    formats(this.#ajv);
    this.#ajv.addMetaSchema(JSON.parse(
      fs.readFileSync(new URL('../../node_modules/ajv/dist/refs/json-schema-draft-07.json', import.meta.url), 'utf8'),
    ));
    this.#ajv.addSchema(JSON.parse(
      fs.readFileSync(new URL('../schemas/definitions.json', import.meta.url), 'utf8'),
    ));

    this.#unsubscribe = uw.events.on(CONFIG_UPDATE_MESSAGE, async (data) => {
      this.#logger.trace({ data }, 'handle config update');

      try {
        const updatedSettings = await this.get(data.key);
        this.#emitter.emit(data.key, updatedSettings, data.user, data.patch);
      } catch (error) {
        this.#logger.error({ err: error }, 'could not retrieve settings after update');
      }
    });
  }

  /**
   * @template {JsonObject} TSettings
   * @param {string} key
   * @param {(settings: TSettings, user: UserID|null, patch: Partial<TSettings>) => void} listener
   */
  subscribe(key, listener) {
    this.#emitter.on(key, listener);
    return () => this.#emitter.off(key, listener);
  }

  /**
   * @param {string} name
   * @param {JsonObject} value
   * @returns {Promise<JsonObject|null>} The old values.
   */
  async #save(name, value) {
    const { db } = this.#uw;

    const previous = await db.transaction().execute(async (tx) => {
      const row = await tx.selectFrom('configuration')
        .select((eb) => json(eb.ref('value')).as('value'))
        .where('name', '=', name)
        .executeTakeFirst();

      await tx.insertInto('configuration')
        .values({ name, value: jsonb(value) })
        .onConflict((oc) => oc.column('name').doUpdateSet({ value: jsonb(value) }))
        .execute();

      return row?.value != null ? fromJson(row.value) : null;
    });

    return previous;
  }

  /**
   * @param {string} key
   * @returns {Promise<JsonObject|null>}
   */
  async #load(key) {
    const { db } = this.#uw;

    const row = await db.selectFrom('configuration')
      .select((eb) => json(eb.ref('value')).as('value'))
      .where('name', '=', key)
      .executeTakeFirst();
    if (row == null || row.value == null) {
      return null;
    }

    return fromJson(row.value);
  }

  /**
   * Add a config group.
   *
   * @param {string} key - The name of the config group.
   * @param {import('ajv').SchemaObject} schema - The JSON schema that the settings must
   *     follow.
   * @public
   */
  register(key, schema) {
    this.#validators.set(key, this.#ajv.compile(schema));
  }

  /**
   * Get the current settings for a config group.
   *
   * @param {string} key
   * @returns {Promise<undefined | JsonObject>}
   *     `undefined` if the config group named `key` does not
   *     exist. An object containing current settings otherwise.
   * @public
   */
  async get(key) {
    const validate = this.#validators.get(key);
    if (!validate) {
      return undefined;
    }

    const config = (await this.#load(key)) ?? {};
    // Allowed to fail--just fills in defaults
    validate(config);

    return config;
  }

  /**
   * Update settings for a config group. Optionally specify the user who is updating the settings.
   *
   * Rejects if the settings do not follow the schema for the config group.
   *
   * @param {string} key
   * @param {JsonObject} settings
   * @param {{ user?: User }} [options]
   * @public
   */
  async set(key, settings, options = {}) {
    const { user } = options;
    const validate = this.#validators.get(key);
    if (validate) {
      if (!validate(settings)) {
        assert(validate.errors, 'Unexpected null');
        this.#logger.trace({ key, errors: validate.errors }, 'config validation error');
        throw new ValidationError(validate.errors, this.#ajv);
      }
    }

    const oldSettings = await this.#save(key, settings);
    const patch = jsonMergePatch.generate(oldSettings, settings) ?? Object.create(null);

    this.#logger.trace({ key, patch }, 'fire config update');
    await this.#uw.publish(CONFIG_UPDATE_MESSAGE, {
      key,
      user: user ? user.id : null,
      patch,
    });
  }

  /**
   * Get *all* settings.
   *
   * @returns {Promise<{ [key: string]: JsonObject }>}
   */
  async getAllConfig() {
    const { db } = this.#uw;

    const results = await db.selectFrom('configuration')
      .select(['name', sql`json(value)`.as('value')])
      .execute();

    const configs = Object.create(null);
    for (const [key, validate] of this.#validators.entries()) {
      const row = results.find((m) => m.name === key);
      if (row) {
        const value = JSON.parse(/** @type {string} */ (row.value));
        validate(value);
        configs[key] = value;
      } else {
        configs[key] = {};
      }
    }
    return configs;
  }

  /**
   * @returns {import('ajv').SchemaObject}
   */
  getSchema() {
    const properties = Object.create(null);
    const required = [];
    for (const [key, validate] of this.#validators.entries()) {
      properties[key] = validate.schema;
      required.push(key);
    }

    return {
      type: 'object',
      properties,
      required,
    };
  }

  destroy() {
    this.#unsubscribe();
  }
}

/**
 * @param {import('../Uwave.js').Boot} uw
 */
async function configStorePlugin(uw) {
  uw.config = new ConfigStore(uw);
  uw.onClose(async () => uw.config.destroy());
}

export default configStorePlugin;
export { ConfigStore };

1	import assert from 'node:assert';	1✔
2	import fs from 'node:fs';	1✔
3	import EventEmitter from 'node:events';	1✔
4	import Ajv from 'ajv/dist/2019.js';	1✔
5	import formats from 'ajv-formats';	1✔
6	import jsonMergePatch from 'json-merge-patch';	1✔
7	import ValidationError from '../errors/ValidationError.js';	1✔
8	import { sql } from 'kysely';	1✔
9	import { fromJson, json, jsonb } from '../utils/sqlite.js';	1✔
10		1✔
11	/**	1✔
12	* @typedef {import('type-fest').JsonObject} JsonObject	1✔
13	* @typedef {import('../schema.js').UserID} UserID	1✔
14	* @typedef {import('../schema.js').User} User	1✔
15	*/	1✔
16		1✔
17	const CONFIG_UPDATE_MESSAGE = 'configStore:update';	1✔
18		1✔
19	/**	1✔
20	* Extensible configuration store.	1✔
21	*	1✔
22	* The config store contains named groups of settings. Each setting group is	1✔
23	* stored in its own MongoDB Document. Groups have associated JSON Schemas to	1✔
24	* check that the configuration is correct.	1✔
25	*/	1✔
26	class ConfigStore {	1✔
27	#uw;	180✔
28		180✔
29	#logger;	180✔
30		180✔
31	#ajv;	180✔
32		180✔
33	#emitter = new EventEmitter();	180✔
34		180✔
35	#unsubscribe;	180✔
36		180✔
37	/** @type {Map<string, import('ajv').ValidateFunction<unknown>>} */	180✔
38	#validators = new Map();	180✔
39		180✔
40	/**	180✔
41	* @param {import('../Uwave.js').Boot} uw	180✔
42	*/	180✔
43	constructor(uw) {	180✔
44	this.#uw = uw;	180✔
45	this.#logger = uw.logger.child({ ns: 'uwave:config' });	180✔
46	this.#ajv = new Ajv({	180✔
47	useDefaults: true,	180✔
48	// Allow unknown keywords (`uw:xyz`)	180✔
49	strict: false,	180✔
50	strictTypes: true,	180✔
51	});	180✔
52	formats(this.#ajv);	180✔
53	this.#ajv.addMetaSchema(JSON.parse(	180✔
54	fs.readFileSync(new URL('../../node_modules/ajv/dist/refs/json-schema-draft-07.json', import.meta.url), 'utf8'),	180✔
55	));	180✔
56	this.#ajv.addSchema(JSON.parse(	180✔
57	fs.readFileSync(new URL('../schemas/definitions.json', import.meta.url), 'utf8'),	180✔
58	));	180✔
59		180✔
60	this.#unsubscribe = uw.events.on(CONFIG_UPDATE_MESSAGE, async (data) => {	180✔
61	this.#logger.trace({ data }, 'handle config update');	14✔
62		14✔
63	try {	14✔
64	const updatedSettings = await this.get(data.key);	14✔
65	this.#emitter.emit(data.key, updatedSettings, data.user, data.patch);	14✔
66	} catch (error) {	14!
NEW 67	this.#logger.error({ err: error }, 'could not retrieve settings after update');	×
NEW 68	}	×
69	});	180✔
70	}	180✔
71		180✔
72	/**	180✔
73	* @template {JsonObject} TSettings	180✔
74	* @param {string} key	180✔
75	* @param {(settings: TSettings, user: UserID\|null, patch: Partial<TSettings>) => void} listener	180✔
76	*/	180✔
77	subscribe(key, listener) {	180✔
78	this.#emitter.on(key, listener);	541✔
79	return () => this.#emitter.off(key, listener);	541✔
80	}	541✔
81		180✔
82	/**	180✔
83	* @param {string} name	180✔
84	* @param {JsonObject} value	180✔
85	* @returns {Promise<JsonObject\|null>} The old values.	180✔
86	*/	180✔
87	async #save(name, value) {	180✔
88	const { db } = this.#uw;	14✔
89		14✔
90	const previous = await db.transaction().execute(async (tx) => {	14✔
91	const row = await tx.selectFrom('configuration')	14✔
92	.select((eb) => json(eb.ref('value')).as('value'))	14✔
93	.where('name', '=', name)	14✔
94	.executeTakeFirst();	14✔
95		14✔
96	await tx.insertInto('configuration')	14✔
97	.values({ name, value: jsonb(value) })	14✔
98	.onConflict((oc) => oc.column('name').doUpdateSet({ value: jsonb(value) }))	14✔
99	.execute();	14✔
100		14✔
101	return row?.value != null ? fromJson(row.value) : null;	14✔
102	});	14✔
103		14✔
104	return previous;	14✔
105	}	14✔
106		180✔
107	/**	180✔
108	* @param {string} key	180✔
109	* @returns {Promise<JsonObject\|null>}	180✔
110	*/	180✔
111	async #load(key) {	180✔
112	const { db } = this.#uw;	456✔
113		456✔
114	const row = await db.selectFrom('configuration')	456✔
115	.select((eb) => json(eb.ref('value')).as('value'))	456✔
116	.where('name', '=', key)	456✔
117	.executeTakeFirst();	456✔
118	if (row == null \|\| row.value == null) {	456✔
119	return null;	427✔
120	}	427✔
121		29✔
122	return fromJson(row.value);	29✔
123	}	456✔
124		180✔
125	/**	180✔
126	* Add a config group.	180✔
127	*	180✔
128	* @param {string} key - The name of the config group.	180✔
129	* @param {import('ajv').SchemaObject} schema - The JSON schema that the settings must	180✔
130	* follow.	180✔
131	* @public	180✔
132	*/	180✔
133	register(key, schema) {	180✔
134	this.#validators.set(key, this.#ajv.compile(schema));	720✔
135	}	720✔
136		180✔
137	/**	180✔
138	* Get the current settings for a config group.	180✔
139	*	180✔
140	* @param {string} key	180✔
141	* @returns {Promise<undefined \| JsonObject>}	180✔
142	* `undefined` if the config group named `key` does not	180✔
143	* exist. An object containing current settings otherwise.	180✔
144	* @public	180✔
145	*/	180✔
146	async get(key) {	180✔
147	const validate = this.#validators.get(key);	458✔
148	if (!validate) {	458✔
149	return undefined;	2✔
150	}	2✔
151		456✔
152	const config = (await this.#load(key)) ?? {};	458✔
153	// Allowed to fail--just fills in defaults	458✔
154	validate(config);	458✔
155		458✔
156	return config;	458✔
157	}	458✔
158		180✔
159	/**	180✔
160	* Update settings for a config group. Optionally specify the user who is updating the settings.	180✔
161	*	180✔
162	* Rejects if the settings do not follow the schema for the config group.	180✔
163	*	180✔
164	* @param {string} key	180✔
165	* @param {JsonObject} settings	180✔
166	* @param {{ user?: User }} [options]	180✔
167	* @public	180✔
168	*/	180✔
169	async set(key, settings, options = {}) {	180✔
170	const { user } = options;	15✔
171	const validate = this.#validators.get(key);	15✔
172	if (validate) {	15✔
173	if (!validate(settings)) {	15✔
174	assert(validate.errors, 'Unexpected null');	1✔
175	this.#logger.trace({ key, errors: validate.errors }, 'config validation error');	1✔
176	throw new ValidationError(validate.errors, this.#ajv);	1✔
177	}	1✔
178	}	15✔
179		14✔
180	const oldSettings = await this.#save(key, settings);	14✔
181	const patch = jsonMergePatch.generate(oldSettings, settings) ?? Object.create(null);	15!
182		15✔
183	this.#logger.trace({ key, patch }, 'fire config update');	15✔
184	await this.#uw.publish(CONFIG_UPDATE_MESSAGE, {	15✔
185	key,	15✔
186	user: user ? user.id : null,	15✔
187	patch,	15✔
188	});	15✔
189	}	15✔
190		180✔
191	/**	180✔
192	* Get all settings.	180✔
193	*	180✔
194	* @returns {Promise<{ [key: string]: JsonObject }>}	180✔
195	*/	180✔
196	async getAllConfig() {	180✔
197	const { db } = this.#uw;	3✔
198		3✔
199	const results = await db.selectFrom('configuration')	3✔
200	.select(['name', sql`json(value)`.as('value')])	3✔
201	.execute();	3✔
202		3✔
203	const configs = Object.create(null);	3✔
204	for (const [key, validate] of this.#validators.entries()) {	3✔
205	const row = results.find((m) => m.name === key);	12✔
206	if (row) {	12!
207	const value = JSON.parse(/** @type {string} */ (row.value));	×
208	validate(value);	×
209	configs[key] = value;	×
210	} else {	12✔
211	configs[key] = {};	12✔
212	}	12✔
213	}	12✔
214	return configs;	3✔
215	}	3✔
216		180✔
217	/**	180✔
218	* @returns {import('ajv').SchemaObject}	180✔
219	*/	180✔
220	getSchema() {	180✔
221	const properties = Object.create(null);	13✔
222	const required = [];	13✔
223	for (const [key, validate] of this.#validators.entries()) {	13✔
224	properties[key] = validate.schema;	52✔
225	required.push(key);	52✔
226	}	52✔
227		13✔
228	return {	13✔
229	type: 'object',	13✔
230	properties,	13✔
231	required,	13✔
232	};	13✔
233	}	13✔
234		180✔
235	destroy() {	180✔
236	this.#unsubscribe();	180✔
237	}	180✔
238	}	180✔
239		1✔
240	/**	1✔
241	* @param {import('../Uwave.js').Boot} uw	1✔
242	*/	1✔
243	async function configStorePlugin(uw) {	180✔
244	uw.config = new ConfigStore(uw);	180✔
245	uw.onClose(async () => uw.config.destroy());	180✔
246	}	180✔
247		1✔
248	export default configStorePlugin;	1✔
249	export { ConfigStore };	1✔

u-wave / core / 23111218316

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous