diff --git a/E2E_USER_STORIES.md b/E2E_USER_STORIES.md index 38e2b28143b..dd7198c9ed3 100644 --- a/E2E_USER_STORIES.md +++ b/E2E_USER_STORIES.md @@ -455,10 +455,11 @@ runs `source` and `mosquitto` for this, and the specs drive state with a purpose `drivers/place/e2e/state_harness.cr` (a UDP device whose `connected` and other statuses are set through `exec`). See "MQTT" in `e2e/README.md`. -**The broker is asserted directly, not only through the page.** Stagehand re-checks every -MQTT alert over REST before showing it (`DashboardsService._updateDashboardAlertList`), which -hides wrong data on the broker. Removing that re-check — it is what makes stagehand slow — is -only safe once the MQTT rows below are all green. +**The broker is asserted directly, not only through the page.** Stagehand builds the hot list +from the broker's retained state alone: the latest value of each topic, with an emptied topic +removed. It used to re-check every alert over REST (a module list and a state read per alert, on +every message), which made it slow and hid wrong data on the broker; that went once the MQTT rows +below were green. | ID | P | Story | Status | |----|---|-------|--------| @@ -479,7 +480,8 @@ only safe once the MQTT rows below are all green. | SH-MQTT-19 | P0 | Two quick edits to one system leave source's mapping matching the latest. | **done** — the lifecycle setup makes them (two modules added back to back). pg-orm and placeos-resource handle each change on its own fiber, so source processed them out of order and kept the older module list. Found while building this coverage. | | SH-MQTT-20 | P1 | After a broker restart, every status is published again. | **done, not automated** — verified by hand (restart mosquitto: 32 of 32 topics back within 2 s; before the fix, 0 until each value changed). | | SH-MQTT-21 | P1 | Retained topics nothing maps to (left by a module deleted while source was down, or by an older source) are removed when source starts. | **done, not automated** — verified by hand with a planted topic. source spec `sweeps retained topics that nothing maps to`. | -| SH-HOT-02 | P1 | A module added to a system after stagehand has loaded it can still raise an alert. | **done** — `local/stagehand-hotlist.spec.ts`. Stagehand cached each system's module list forever (`_system_modules_cache`); it now refetches, at most every 5 s, before dropping an alert for an unknown device. | +| SH-HOT-02 | P1 | A module added to a system after stagehand has loaded it can still raise an alert. | **done** — `local/stagehand-hotlist.spec.ts`. Was a module list cached forever by the REST re-check, which is gone. | +| SH-HOT-03 | P0 | A deleted module leaves the hot list, the module after it is listed at its index, and the page makes no per-alert REST calls. | **done** — `local/stagehand-hotlist.spec.ts`. Red against the REST re-checking stagehand: an emptied topic failed to parse, so the deleted module stayed listed. | **The lifecycle specs need source v1.22.1 or later** (PlaceOS/source#65). Against `placeos-2.2610.1` SH-MQTT-10 … 19 and SH-HOT-02 are red, which is what they are for. Until a diff --git a/apps/stagehand/e2e/local/stagehand-hotlist.spec.ts b/apps/stagehand/e2e/local/stagehand-hotlist.spec.ts index 50eb47acf17..a923c54c24b 100644 --- a/apps/stagehand/e2e/local/stagehand-hotlist.spec.ts +++ b/apps/stagehand/e2e/local/stagehand-hotlist.spec.ts @@ -6,9 +6,9 @@ * harness reports connected: false -> listed * harness reports connected: true -> gone again * - * The broker is asserted alongside the page at every step. When stagehand - * stops re-checking alerts over REST, this spec is what says the page is still - * right; until then, the broker assertions are what say the data is. + * The broker is asserted alongside the page at every step. Stagehand takes the + * hot list straight from the broker's retained state, with no REST checks, so + * these are what say the page follows the broker. * * "Not listed" needs a control (see e2e/README.md, Conventions): the page shows * a spinner, not an empty table, until its first alert. So the system also @@ -29,6 +29,7 @@ import { expect, test } from '../../../../e2e/support/stagehand/fixtures'; import { addHarness, createSystem, + deleteModule, deleteSystem, setConnected, sweepSystems, @@ -84,10 +85,6 @@ test('a module is listed while it reports connected: false, and only then', asyn MQTT_PROPAGATION_MS, ); - // NOTE: both modules exist before the system's first alert. Stagehand - // caches a system's module list on its first alert and never refreshes - // it, so a module added after that cannot alert until a reload - // (SH-HOT-02). This spec is about the working path. await setConnected(adminApi, control.id, false); await broker.waitForValue( 'Projector', @@ -194,3 +191,65 @@ test('SH-HOT-02: a module added after the page loaded the system can still be li await deleteSystem(adminApi, system.id); } }); + +test('SH-HOT-03: a deleted module leaves the list, and the module after it takes its index, without REST checks', async ({ + page, + adminApi, + stagehandAdminToken, + harnessDriverId, + zones, +}) => { + const system = await createSystem(adminApi, zones, 'delete-listed'); + const broker = await BrokerView.open( + stagehandAdminToken, + systemFilter(zones.org, system.id), + ); + // the per-alert checks stagehand used to make: a system's modules, and a + // module's state + const rest_checks: string[] = []; + page.on('request', (r) => { + const url = r.url(); + if ( + /\/api\/engine\/v2\/modules\?.*control_system_id=/.test(url) || + new RegExp(`/api/engine/v2/systems/${system.id}/[^/?]+_\\d+`).test( + url, + ) + ) { + rest_checks.push(`${r.method()} ${url}`); + } + }); + try { + await openHotList(page); + + const first = await addHarness(adminApi, system.id, harnessDriverId); + const second = await addHarness(adminApi, system.id, harnessDriverId); + await waitForConnected(adminApi, first.id, true); + await waitForConnected(adminApi, second.id, true); + await setConnected(adminApi, first.id, false); + await setConnected(adminApi, second.id, false); + await expectHotList(page, system.id, ['Display_1', 'Display_2']); + + // the second becomes Display_1; nothing is left at Display_2 + await deleteModule(adminApi, first.id); + await expectHotList(page, system.id, ['Display_1']); + + // and it is the second module listed there, so reconnecting it clears it + await setConnected(adminApi, second.id, true); + await broker.waitForValue( + 'Display', + 1, + 'connected', + true, + MQTT_PROPAGATION_MS, + ); + await expectHotList(page, system.id, []); + + expect(rest_checks).toEqual([]); + } finally { + await test + .info() + .attach('broker messages', { body: broker.describeLog() }); + await broker.close(); + await deleteSystem(adminApi, system.id); + } +}); diff --git a/apps/stagehand/src/app/alerts.service.ts b/apps/stagehand/src/app/alerts.service.ts deleted file mode 100644 index 64001b5c40b..00000000000 --- a/apps/stagehand/src/app/alerts.service.ts +++ /dev/null @@ -1,135 +0,0 @@ -import { inject, Injectable, signal } from '@angular/core'; -import { OrganisationService } from '@placeos/common'; -import { token } from '@placeos/ts-client'; - -import mqtt from 'mqtt'; - -type AlertSeverity = 'critical' | 'high' | 'medium' | 'low'; -type AlertStatus = 'open' | 'in progress' | 'closed' | 'resolved'; -type DeviceType = 'display' | 'audio' | 'video' | 'network' | 'control'; - -export interface Alert { - id: string; - severity: AlertSeverity; - type: DeviceType; - location: string; - device: string; - subject: string; - body: string; - status: AlertStatus; - timestamp: number; -} - -export interface StateTopic { - building_id: string; - level_id: string; - area_id: string; - system_id: string; - driver_id: string; - module_name: string; - module_index: string; - state_key: string; -} - -function stringToTopic(topic: string): StateTopic { - const [ - _, - _o, - _s, - bld_id, - lvl_id, - area_id, - _b, - sys_id, - dev_id, - mod_id, - mod_idx, - key, - ] = topic.split('/'); - return { - building_id: bld_id, - level_id: lvl_id, - area_id: area_id, - system_id: sys_id, - driver_id: dev_id, - module_name: mod_id, - module_index: mod_idx, - state_key: key, - }; -} - -@Injectable({ - providedIn: 'root', -}) -export class AlertsService { - private _org = inject(OrganisationService); - private _broker: mqtt.MqttClient; - - public readonly system_state = signal({}); - public readonly alerts = signal([]); - - constructor() { - this.init(); - } - - public async init() { - await this._org.waitUntilInitialised(); - const jwt = token(); - const url = `${location.host}/api/mqtt/`; - const secure = location.protocol === 'https:'; - this._broker = mqtt.connect(`ws${secure ? 's' : ''}://${url}`, { - username: jwt, - password: jwt, - }); - this._broker.on('connect', () => { - this._broker.subscribe( - `placeos/${this._org.organisation.id}/state/+/+/+/+/+/+/+/+/connected`, - ); - }); - this._broker.on('message', (t, msg) => { - const data = JSON.parse(msg.toString()); - const topic = stringToTopic(t); - console.log('MQTT:', topic, data); - if (topic.state_key === 'connected') { - this.system_state.update((old) => { - const sys = old[topic.system_id] || { - connected: [], - disconnected: [], - }; - console.log('System:', sys); - const mod = `${topic.module_name}_${topic.module_index}`; - sys.connected = sys.connected.filter((_) => _ !== mod); - sys.disconnected = sys.disconnected.filter( - (_) => _ !== mod, - ); - data.value === true - ? sys.connected.push(mod) - : sys.disconnected.push(mod); - this.alerts.update((list) => { - const id = `${topic.system_id}+${topic.module_name}_${topic.module_index}`; - return data.value - ? [...list.filter((_) => _.id !== id)] - : [ - ...list.filter((_) => _.id !== id), - { - id, - severity: 'high', - type: 'control', - subject: 'Disconnected', - location: topic.system_id, - device: ``, - body: `Module "${topic.module_name}_${topic.module_index}" is disconnected`, - timestamp: data.timestamp, - status: 'open', - }, - ]; - }); - return { - ...old, - [topic.system_id]: sys, - }; - }); - } - }); - } -} diff --git a/apps/stagehand/src/app/dashboards/broker-alerts.ts b/apps/stagehand/src/app/dashboards/broker-alerts.ts new file mode 100644 index 00000000000..a1561f23c56 --- /dev/null +++ b/apps/stagehand/src/app/dashboards/broker-alerts.ts @@ -0,0 +1,261 @@ +import { unique } from '@placeos/common'; +import { + PlaceAlert, + TriggerComparison, + TriggerConditionOperator, +} from '@placeos/ts-client'; + +type AlertSeverity = 'critical' | 'high' | 'medium' | 'low'; +type AlertStatus = 'open' | 'in progress' | 'closed' | 'resolved'; +type DeviceType = 'display' | 'audio' | 'video' | 'network' | 'control'; + +/** An alert raised on the hot list */ +export interface Alert { + id: string; + severity: AlertSeverity; + type: DeviceType; + location: string; + device: string; + subject: string; + body: string; + status: AlertStatus; + timestamp: number; +} + +export interface StateTopic { + region_id?: string; + building_id?: string; + level_id?: string; + area_id?: string; + system_id?: string; + driver_id?: string; + module_name?: string; + module_index?: string; + state_key?: string; +} + +/** The latest value the broker holds for one status topic */ +export interface BrokerValue { + /** epoch seconds, from source */ + time: number; + topic: StateTopic; + topic_str: string; + value: unknown; +} + +export function topicToString(topic: StateTopic, org_id = ''): string { + return `placeos/${org_id || '+'}/+/${topic.region_id || '+'}/${topic.building_id || '+'}/${topic.level_id || '+'}/${topic.area_id || '+'}/${topic.system_id || '+'}/${topic.driver_id || '+'}/${topic.module_name || '+'}/${topic.module_index || '+'}/${topic.state_key || '+'}`; +} + +export function stringToTopic(topic: string): StateTopic { + const [ + _, + _o, + _s, + bld_id, + lvl_id, + area_id, + _b, + sys_id, + dev_id, + mod_id, + mod_idx, + key, + ] = topic.split('/'); + return { + building_id: bld_id, + level_id: lvl_id, + area_id: area_id, + system_id: sys_id, + driver_id: dev_id, + module_name: mod_id, + module_index: mod_idx, + state_key: key, + }; +} + +/** The topic filter a comparison listens on */ +export function compareAsTopic( + item: TriggerComparison, + zones: { region?: string; building?: string; level?: string } = {}, + org_id = '', +): string { + if (!(item.left instanceof Object)) + return topicToString( + { + state_key: `${item.left}`, + region_id: zones.region, + building_id: zones.building, + }, + org_id, + ); + const parts = (item.left?.mod || '').split('_'); + parts.pop(); + const module_name = parts.join('_'); + return topicToString( + { + module_name, + state_key: item.left.status, + region_id: zones.region, + building_id: zones.building, + level_id: zones.level, + }, + org_id, + ); +} + +export function matchTopics(topic: string, comp: string): boolean { + const t_state = topic.split('/'); + const comp_t_state = comp.split('/'); + for (let i = 0; i < t_state.length; i++) { + if ( + t_state[i] !== comp_t_state[i] && + t_state[i] !== '+' && + comp_t_state[i] !== '+' + ) { + return false; + } + } + return true; +} + +/** + * Reads a broker message. `null` means the topic was removed: source clears a + * retained topic by publishing an empty payload to it, when a status is + * cleared or the module it belonged to is gone. + */ +export function parseBrokerMessage( + topic_str: string, + payload: string, +): BrokerValue | null { + if (!payload) return null; + const data = JSON.parse(payload); + return { + time: data.time, + value: data.value, + topic: stringToTopic(topic_str), + topic_str, + }; +} + +/** "true", "12" and "null" as the values they stand for */ +export function coerceValue(value: unknown): unknown { + if (typeof value !== 'string') return value; + const lower = value.toLowerCase(); + if (lower === 'true') return true; + if (lower === 'false') return false; + if (lower === 'null') return null; + if (lower === 'undefined') return undefined; + if (value !== '' && !isNaN(Number(value))) return Number(value); + return value; +} + +/** The comparison's left-hand value, read from a status value */ +function comparedValue(value: unknown, item: TriggerComparison): unknown { + let out = value; + const keys = item.left instanceof Object ? item.left.keys || [] : []; + for (const key of keys) { + if (out == null) break; + out = out[key]; + } + return out; +} + +/** Whether `value expected` holds */ +export function evaluate( + value: unknown, + operator: TriggerConditionOperator, + expected: unknown, +): boolean { + // compared as JavaScript compares them, as the REST checks did + const current = coerceValue(value) as number; + const right = coerceValue(expected) as number; + switch (operator) { + case TriggerConditionOperator.EQ: + return current === right; + case TriggerConditionOperator.NEQ: + return current !== right; + case TriggerConditionOperator.GT: + return current > right; + case TriggerConditionOperator.LT: + return current < right; + case TriggerConditionOperator.GTE: + return current >= right; + case TriggerConditionOperator.LTE: + return current <= right; + case TriggerConditionOperator.AND: + return !!(current && right); + case TriggerConditionOperator.OR: + return !!(current || right); + case TriggerConditionOperator.XOR: + return !!current !== !!right; + default: + return false; + } +} + +/** The broker values a comparison holds for */ +export function findCompareMatches( + values: Iterable, + item: TriggerComparison, +): BrokerValue[] { + const topic = compareAsTopic(item); + const matches: BrokerValue[] = []; + for (const entry of values) { + if (!matchTopics(topic, entry.topic_str)) continue; + const value = comparedValue(entry.value, item); + if (evaluate(value, item.operator, item.right)) matches.push(entry); + } + return matches; +} + +/** + * The alerts raised by the broker's current values. + * + * NOTE:: `values` holds only the latest value of each topic, and only topics + * the broker still retains. Source keeps the broker matching the modules, + * their order and their state, so nothing here is checked again over REST. + */ +export function raisedAlerts( + definitions: PlaceAlert[], + values: Iterable, +): Alert[] { + const list = [...values]; + let out: Alert[] = []; + for (const alert of definitions) { + const comparisons = alert.conditions?.comparisons || []; + if (!comparisons.length) continue; + const comparison_matches = comparisons.map((item) => + findCompareMatches(list, item), + ); + const required = alert.any_match ? 1 : comparisons.length; + for (const match of comparison_matches[0]) { + const group = [match]; + for (let i = 1; i < comparison_matches.length; i++) { + // conditions are about one system, within the debounce period + const other = comparison_matches[i].find( + (_) => + _.topic.system_id === match.topic.system_id && + Math.abs(_.time - match.time) < alert.debounce_period, + ); + if (other) group.push(other); + } + if (group.length < required) continue; + const { time, topic } = group[0]; + const device = `${topic.module_name}_${topic.module_index}`; + out.push({ + id: `${alert.id}+${topic.system_id}+${device}`, + severity: alert.severity as Alert['severity'], + type: alert.alert_type as Alert['type'], + location: topic.system_id, + device, + subject: alert.name, + body: alert.description, + status: 'open', + timestamp: time * 1000, + }); + } + out = unique(out, 'id'); + } + return out; +} diff --git a/apps/stagehand/src/app/dashboards/dashboards.service.ts b/apps/stagehand/src/app/dashboards/dashboards.service.ts index e4e9c130bc0..adcf59a5170 100644 --- a/apps/stagehand/src/app/dashboards/dashboards.service.ts +++ b/apps/stagehand/src/app/dashboards/dashboards.service.ts @@ -6,196 +6,30 @@ import { log, notifySuccess, OrganisationService, - unique, } from '@placeos/common'; import { listDashboardAlerts, PlaceAlert, PlaceAlertDashboard, - PlaceModule, queryAlertDashboards, - queryModules, removeAlert, removeAlertDashboard, showAlert, showAlertDashboard, - systemModuleState, token, - TriggerComparison, TriggerConditionOperator, } from '@placeos/ts-client'; import { openConfirmModal } from '@placeos/components'; import mqtt from 'mqtt'; -import { Alert } from '../alerts.service'; import { AlertNotificationService } from '../push-notification.service'; - -export interface StateTopic { - region_id?: string; - building_id?: string; - level_id?: string; - area_id?: string; - system_id?: string; - driver_id?: string; - module_name?: string; - module_index?: string; - state_key?: string; -} - -let _org_id = ''; - -function topicToString(topic: StateTopic): string { - return `placeos/${_org_id || '+'}/+/${topic.region_id || '+'}/${topic.building_id || '+'}/${topic.level_id || '+'}/${topic.area_id || '+'}/${topic.system_id || '+'}/${topic.driver_id || '+'}/${topic.module_name || '+'}/${topic.module_index || '+'}/${topic.state_key || '+'}`; -} - -function stringToTopic(topic: string): StateTopic { - const [ - _, - _o, - _s, - bld_id, - lvl_id, - area_id, - _b, - sys_id, - dev_id, - mod_id, - mod_idx, - key, - ] = topic.split('/'); - return { - building_id: bld_id, - level_id: lvl_id, - area_id: area_id, - system_id: sys_id, - driver_id: dev_id, - module_name: mod_id, - module_index: mod_idx, - state_key: key, - }; -} - -function compareAsTopic( - item: TriggerComparison, - zones: { region?: string; building?: string; level?: string } = {}, -): string { - if (!(item.left instanceof Object)) - return topicToString({ - state_key: `${item.left}`, - region_id: zones.region, - building_id: zones.building, - }); - const parts = (item.left?.mod || '').split('_'); - parts.pop(); - const module_name = parts.join('_'); - const topic = topicToString({ - module_name, - state_key: item.left.status, - region_id: zones.region, - building_id: zones.building, - level_id: zones.level, - }); - return topic; -} - -function matchTopics(topic: string, comp: string): boolean { - const t_state = topic.split('/'); - const comp_t_state = comp.split('/'); - for (let i = 0; i < t_state.length; i++) { - if ( - t_state[i] !== comp_t_state[i] && - t_state[i] !== '+' && - comp_t_state[i] !== '+' - ) { - return false; - } - } - return true; -} - -/** Minimum time between refetches of a system's modules on an unknown device */ -const MODULE_LIST_REFRESH_MS = 5000; - -interface BrokerAlert { - time: number; - topic: StateTopic; - topic_str: string; - value: any; -} - -/** - * Calculate module index based on modules of the same name/class - */ -function calculateModuleIndex( - module_list: PlaceModule[], - module: PlaceModule, -): number { - const driver = module.driver || { class_name: 'System' }; - const module_class = module.custom_name || module.name || driver.class_name; - const modules_with_class = module_list.filter((mod) => { - const d = mod.driver || { class_name: 'System' }; - const mod_class = mod.custom_name || mod.name || d.class_name; - return mod_class === module_class; - }); - return Math.max( - 1, - modules_with_class.findIndex((mod) => mod.id === module.id) + 1, - ); -} - -/** - * Get list of valid device names (module_name_index format) for a system's modules - */ -function getModuleDeviceNames(modules: PlaceModule[]): string[] { - return modules.map((mod) => { - const driver = mod.driver || { class_name: 'System' }; - const name = - mod.custom_name || mod.name || driver.class_name || 'Blank'; - const index = calculateModuleIndex(modules, mod); - return `${name}_${index}`; - }); -} - -function findCompareMatches(alerts: BrokerAlert[], item: TriggerComparison) { - const topic = compareAsTopic(item); - const matches: BrokerAlert[] = []; - for (const alert of alerts) { - if (matchTopics(topic, alert.topic_str)) { - switch (item.operator) { - case TriggerConditionOperator.EQ: - item.right === alert.value ? matches.push(alert) : ''; - break; - case TriggerConditionOperator.NEQ: - item.right !== alert.value ? matches.push(alert) : ''; - break; - case TriggerConditionOperator.GT: - item.right > alert.value ? matches.push(alert) : ''; - break; - case TriggerConditionOperator.LT: - item.right < alert.value ? matches.push(alert) : ''; - break; - case TriggerConditionOperator.GTE: - item.right >= alert.value ? matches.push(alert) : ''; - break; - case TriggerConditionOperator.LTE: - item.right <= alert.value ? matches.push(alert) : ''; - break; - case TriggerConditionOperator.AND: - item.right && alert.value ? matches.push(alert) : ''; - break; - case TriggerConditionOperator.OR: - item.right || alert.value ? matches.push(alert) : ''; - break; - case TriggerConditionOperator.XOR: - (item.right as number) | alert.value - ? matches.push(alert) - : ''; - break; - } - } - } - return matches; -} +import { + Alert, + BrokerValue, + compareAsTopic, + parseBrokerMessage, + raisedAlerts, +} from './broker-alerts'; @Injectable({ providedIn: 'root', @@ -207,7 +41,8 @@ export class DashboardsService extends AsyncHandler { private _mqtt_broker: mqtt.MqttClient; private _connected = false; - private _alerts = signal([]); + /** The latest value of each topic the broker holds, by topic */ + private _values = new Map(); private _initialising = signal(false); private _region_set_from_params = false; /** Capture original URL query string at construction time before effects can modify it */ @@ -218,12 +53,6 @@ export class DashboardsService extends AsyncHandler { })(); /** Track alert IDs that have already been notified to prevent duplicates */ private _notified_alert_ids = new Set(); - /** Cache of system modules: system_id -> list of valid device names */ - private _system_modules_cache = new Map(); - /** When each system's module list was fetched, to bound refetches */ - private _system_modules_fetched_at = new Map(); - /** Track in-flight module queries to prevent duplicate requests */ - private _pending_module_queries = new Map>(); /** Track current alert subscription scope to avoid unnecessary resets */ private _alert_scope = ''; @@ -386,7 +215,7 @@ export class DashboardsService extends AsyncHandler { if (scope_changed) { this.dashboard_alerts.set([]); - this._alerts.set([]); + this._values.clear(); } this._alert_scope = alert_scope; if (!this._mqtt_broker) await this._initialiseBroker(); @@ -399,7 +228,6 @@ export class DashboardsService extends AsyncHandler { try { await this._org.waitUntilInitialised(); this.unsubWith('alert:'); - _org_id = this._org.organisation.id; const jwt = token(); const url = `${location.host}/api/mqtt/`; const secure = location.protocol === 'https:'; @@ -411,7 +239,13 @@ export class DashboardsService extends AsyncHandler { keepalive: 30, }, ); - this._mqtt_broker.on('connect', () => (this._connected = true)); + this._mqtt_broker.on('connect', () => { + // on a reconnection the client subscribes again and is handed + // every retained value; a topic removed while it was away + // sends nothing, so start from what it is handed + this._values.clear(); + this._connected = true; + }); this._mqtt_broker.on('disconnect', () => { this._connected = false; delete this._mqtt_broker; @@ -426,200 +260,6 @@ export class DashboardsService extends AsyncHandler { this.timeout('init_finish', () => this._initialising.set(false)); } - /** - * Get valid module device names for a system. - * Caches results to avoid repeated API calls. - */ - private async _getSystemModuleNames( - system_id: string, - refresh = false, - ): Promise { - // Return cached result if available - if (!refresh && this._system_modules_cache.has(system_id)) { - return this._system_modules_cache.get(system_id); - } - - // Check if there's already a pending query for this system - if (this._pending_module_queries.has(system_id)) { - return this._pending_module_queries.get(system_id); - } - - // Create new query promise - const query_promise = queryModules({ control_system_id: system_id }) - .then((resp) => resp.data) - .then((modules) => { - const device_names = getModuleDeviceNames(modules); - this._system_modules_cache.set(system_id, device_names); - this._system_modules_fetched_at.set(system_id, Date.now()); - this._pending_module_queries.delete(system_id); - return device_names; - }) - .catch((err) => { - log( - 'ALERTS', - `Failed to fetch modules for system ${system_id}:`, - err, - ); - this._pending_module_queries.delete(system_id); - // Return empty array on error - this will cause the alert to be filtered out - return []; - }); - - this._pending_module_queries.set(system_id, query_promise); - return query_promise; - } - - /** - * Check if a module exists on a system. - * Returns true if the module exists, false otherwise. - */ - private async _moduleExistsOnSystem( - system_id: string, - device: string, - ): Promise { - const module_names = await this._getSystemModuleNames(system_id); - if (module_names.includes(device)) return true; - // The cached list predates a module added to the system since; check - // again before dropping the alert, but not more often than this - const fetched_at = this._system_modules_fetched_at.get(system_id) ?? 0; - const wait = fetched_at + MODULE_LIST_REFRESH_MS - Date.now(); - if (wait > 0) { - // Nothing may arrive to prompt another look, so schedule one - this.timeout( - 'module_list_recheck', - () => this._updateDashboardAlertList(), - wait + 50, - ); - return false; - } - const refreshed = await this._getSystemModuleNames(system_id, true); - return refreshed.includes(device); - } - - /** - * Get the current value of a status variable from a module. - * Returns undefined if the value cannot be fetched. - */ - private async _getModuleStatusValue( - system_id: string, - device: string, - status_key: string, - keys: string[] = [], - ): Promise { - const parts = (device || '').split('_'); - const module_index = parseInt(parts.pop(), 10) || 1; - const module_name = parts.join('_'); - if (!module_name) return undefined; - - try { - const state = await systemModuleState( - system_id, - module_name, - module_index, - ); - let value = state?.[status_key]; - // Navigate through sub-keys if specified - for (const key of keys || []) { - if (value == null) break; - value = value[key]; - } - return value; - } catch (err) { - log( - 'ALERTS', - `Failed to fetch status value for ${device}.${status_key} on system ${system_id}:`, - err, - ); - return undefined; - } - } - - /** - * Coerce a value that may be a string representation to its actual type. - * Handles "true"/"false" strings, numeric strings, and null/undefined strings. - */ - private _coerceValue(value: unknown): unknown { - if (typeof value !== 'string') return value; - - const lower = value.toLowerCase(); - // Handle boolean strings - if (lower === 'true') return true; - if (lower === 'false') return false; - // Handle null/undefined strings - if (lower === 'null') return null; - if (lower === 'undefined') return undefined; - // Handle numeric strings - if (value !== '' && !isNaN(Number(value))) { - return Number(value); - } - return value; - } - - /** - * Validate that an alert condition matches the current module state. - * This prevents false positives from delayed or stale MQTT messages. - */ - private async _validateAlertCondition( - system_id: string, - device: string, - comparison: TriggerComparison, - ): Promise { - // Get the status key and subkeys from the comparison - let status_key: string; - let keys: string[] = []; - if (comparison.left instanceof Object) { - status_key = comparison.left.status; - keys = comparison.left.keys || []; - } else { - status_key = comparison.left as string; - } - - // Fetch the current value from the module - const raw_value = await this._getModuleStatusValue( - system_id, - device, - status_key, - keys, - ); - - if (raw_value === undefined) { - // If we can't fetch the value, don't validate (let alert through) - log( - 'ALERTS', - `Could not fetch current value for ${device}.${status_key}, skipping validation`, - ); - return true; - } - - // Coerce values that may be string representations to their actual types - const current_value = this._coerceValue(raw_value); - const expected = this._coerceValue(comparison.right); - - // Validate against the comparison operator - switch (comparison.operator) { - case TriggerConditionOperator.EQ: - return current_value === expected; - case TriggerConditionOperator.NEQ: - return current_value !== expected; - case TriggerConditionOperator.GT: - return current_value > expected; - case TriggerConditionOperator.LT: - return current_value < expected; - case TriggerConditionOperator.GTE: - return current_value >= expected; - case TriggerConditionOperator.LTE: - return current_value <= expected; - case TriggerConditionOperator.AND: - return !!(current_value && expected); - case TriggerConditionOperator.OR: - return !!(current_value || expected); - case TriggerConditionOperator.XOR: - return !!current_value !== !!expected; - default: - return true; - } - } - private _listenToAlertTopics() { if (!this._connected) { return this.timeout('alert_topics', () => @@ -627,6 +267,9 @@ export class DashboardsService extends AsyncHandler { ); } this.unsubWith('alert:'); + // resubscribing hands back every retained value, so drop what came + // from the old subscriptions rather than keep values nothing updates + this._values.clear(); const dashboard = this.dashboard(); const alert_list = this.alerts_list(); if (!alert_list.length) { @@ -653,6 +296,7 @@ export class DashboardsService extends AsyncHandler { : this.region_id() ? { region: this.region_id() } : {}, + this._org.organisation.id, ); if (!topic) continue; log('ALERTS', 'Listening to topic:', topic); @@ -669,120 +313,33 @@ export class DashboardsService extends AsyncHandler { } private _handleBrokerMessage(t: string, msg_buffer: Buffer) { - const topic = stringToTopic(t); - const data = JSON.parse(msg_buffer.toString()); - const details = { ...data, topic, topic_str: t }; - this._alerts.update((l) => [...l.filter((_) => _), details]); + let entry: BrokerValue | null; + try { + entry = parseBrokerMessage(t, msg_buffer.toString()); + } catch (e) { + log('ALERTS', `Ignoring unreadable message on ${t}:`, e); + return; + } + // keep only the latest value of each topic, and forget removed ones + if (entry) this._values.set(t, entry); + else this._values.delete(t); this._updateDashboardAlertList(); } private _updateDashboardAlertList() { + // a subscription is handed every retained topic at once, so wait for + // the burst to finish before working the alerts out this.timeout( 'update_alerts', - async () => { - const alert_list = this.alerts_list(); - // Track alerts with their triggering comparisons for validation - let alert_out: (Alert & { - _comparisons?: TriggerComparison[]; - })[] = []; - for (const alert of alert_list) { - const comparision_matches: any[][] = []; - for (const item of alert.conditions.comparisons) { - const matches = findCompareMatches( - this._alerts(), - item, - ); - comparision_matches.push(matches); - } - console.log('Matches:', alert, comparision_matches); - if (!comparision_matches.length) continue; - const match_groups: any[][] = []; - for (const match of comparision_matches[0]) { - const group = [match]; - for (let i = 1; i < comparision_matches.length; i++) { - const other_match = comparision_matches[i].find( - (_) => - Math.abs(_.time - match.time) < - alert.debounce_period, - ); - if (other_match) group.push(other_match); - } - match_groups.push(group); - } - for (const group of match_groups) { - const required_matches = alert.any_match - ? 1 - : alert.conditions.comparisons.length; - if (group.length >= required_matches) { - const { time, topic_str } = group[0]; - const topic = stringToTopic(topic_str); - const device = `${topic.module_name}_${topic.module_index}`; - alert_out.push({ - id: `${alert.id}+${topic.system_id}+${device}`, - severity: alert.severity, - type: alert.alert_type as any, - location: topic.system_id, - device, - subject: alert.name, - body: alert.description, - status: 'open', - timestamp: time * 1000, - // Store comparisons for validation - _comparisons: alert.conditions.comparisons, - }); - } - } - alert_out = unique(alert_out, 'id'); - } - - // Filter out alerts where the module no longer exists on the system - // and validate that current state matches the alert condition - const filtered_alerts: Alert[] = []; - for (const alert of alert_out) { - const exists = await this._moduleExistsOnSystem( - alert.location, - alert.device, - ); - if (!exists) { - log( - 'ALERTS', - `Filtering out alert for non-existent module: ${alert.device} on system ${alert.location}`, - ); - continue; - } - - // Validate current state matches at least one comparison - const comparisons = alert._comparisons || []; - let valid = comparisons.length === 0; // If no comparisons, consider valid - for (const comparison of comparisons) { - const is_valid = await this._validateAlertCondition( - alert.location, - alert.device, - comparison, - ); - if (is_valid) { - valid = true; - break; // At least one comparison is still valid - } - } - - if (valid) { - // Remove internal _comparisons before adding to output - const { _comparisons, ...clean_alert } = alert; - filtered_alerts.push(clean_alert); - } else { - log( - 'ALERTS', - `Filtering out alert for ${alert.device} on system ${alert.location}: current state no longer matches condition`, - ); - } - } - - // Send push notifications for new alerts - this._sendPushNotifications(filtered_alerts); - this.dashboard_alerts.set(filtered_alerts); + () => { + const alerts = raisedAlerts( + this.alerts_list(), + this._values.values(), + ); + this._sendPushNotifications(alerts); + this.dashboard_alerts.set(alerts); }, - 10, + 50, ); } diff --git a/apps/stagehand/src/tests/alerts.service.spec.ts b/apps/stagehand/src/tests/alerts.service.spec.ts deleted file mode 100644 index 6b0101afb70..00000000000 --- a/apps/stagehand/src/tests/alerts.service.spec.ts +++ /dev/null @@ -1,159 +0,0 @@ -import { - createServiceFactory, - SpectatorService, -} from '@ngneat/spectator/vitest'; -import { OrganisationService } from '@placeos/common'; -import { token } from '@placeos/ts-client'; -import { MockProvider } from 'ng-mocks'; - -import { AlertsService } from '../app/alerts.service'; - -const { mock_mqtt_handlers, mock_mqtt_client, mock_mqtt_connect } = vi.hoisted( - () => { - const handlers: Record void> = {}; - const client = { - on: vi.fn((event: string, cb: (...args: unknown[]) => void) => { - handlers[event] = cb; - return client; - }), - subscribe: vi.fn(), - unsubscribe: vi.fn(), - end: vi.fn(), - }; - const connect = vi.fn(() => client); - return { - mock_mqtt_handlers: handlers, - mock_mqtt_client: client, - mock_mqtt_connect: connect, - }; - }, -); - -vi.mock('mqtt', () => ({ - default: { connect: mock_mqtt_connect }, -})); - -vi.mock('@placeos/ts-client', { spy: true }); - -/** Flush the chained awaits inside AlertsService.init() */ -const flush = async () => { - for (let i = 0; i < 8; i++) await Promise.resolve(); -}; - -/** Build a broker topic string as emitted by PlaceOS state channels */ -const state_topic = (system: string, module: string, index: string) => - `placeos/org-1/state/bld-1/lvl-1/area-1/b/${system}/drv-1/${module}/${index}/connected`; - -const message = (value: unknown, timestamp = 111) => - Buffer.from(JSON.stringify({ value, timestamp })); - -describe('AlertsService', () => { - let spectator: SpectatorService; - - const create_service = createServiceFactory({ - service: AlertsService, - providers: [ - MockProvider(OrganisationService, { - waitUntilInitialised: () => Promise.resolve(), - organisation: { id: 'org-1' } as any, - }), - ], - }); - - beforeEach(() => { - vi.mocked(token).mockReturnValue('jwt-token'); - for (const key of Object.keys(mock_mqtt_handlers)) - delete mock_mqtt_handlers[key]; - mock_mqtt_connect.mockClear(); - mock_mqtt_client.subscribe.mockClear(); - mock_mqtt_client.on.mockClear(); - spectator = create_service(); - }); - - it('should create the service and connect to the broker after org init', async () => { - expect(spectator.service).toBeTruthy(); - await flush(); - expect(mock_mqtt_connect).toHaveBeenCalledTimes(1); - const [, options] = mock_mqtt_connect.mock.calls[0] as any; - expect(options).toEqual( - expect.objectContaining({ - username: 'jwt-token', - password: 'jwt-token', - }), - ); - }); - - it('should subscribe to the organisation state topic on connect', async () => { - await flush(); - mock_mqtt_handlers['connect'](); - expect(mock_mqtt_client.subscribe).toHaveBeenCalledWith( - 'placeos/org-1/state/+/+/+/+/+/+/+/+/connected', - ); - }); - - it('should add an alert and mark the module disconnected on a disconnect message', async () => { - await flush(); - mock_mqtt_handlers['message']( - state_topic('sys-1', 'Display', '1'), - message(false), - ); - - const alerts = spectator.service.alerts(); - expect(alerts).toHaveLength(1); - expect(alerts[0]).toEqual( - expect.objectContaining({ - id: 'sys-1+Display_1', - severity: 'high', - type: 'control', - subject: 'Disconnected', - location: 'sys-1', - status: 'open', - }), - ); - expect(alerts[0].body).toContain('Display_1'); - - const state = spectator.service.system_state(); - expect(state['sys-1'].disconnected).toContain('Display_1'); - expect(state['sys-1'].connected).not.toContain('Display_1'); - }); - - it('should clear a prior alert when the module reconnects', async () => { - await flush(); - mock_mqtt_handlers['message']( - state_topic('sys-1', 'Display', '1'), - message(false), - ); - expect(spectator.service.alerts()).toHaveLength(1); - - mock_mqtt_handlers['message']( - state_topic('sys-1', 'Display', '1'), - message(true), - ); - - expect(spectator.service.alerts()).toHaveLength(0); - const state = spectator.service.system_state(); - expect(state['sys-1'].connected).toContain('Display_1'); - expect(state['sys-1'].disconnected).not.toContain('Display_1'); - }); - - it('should track disconnects for independent modules separately', async () => { - await flush(); - mock_mqtt_handlers['message']( - state_topic('sys-1', 'Display', '1'), - message(false), - ); - mock_mqtt_handlers['message']( - state_topic('sys-1', 'Audio', '2'), - message(false), - ); - - const ids = spectator.service.alerts().map((a) => a.id); - expect(ids).toEqual( - expect.arrayContaining(['sys-1+Display_1', 'sys-1+Audio_2']), - ); - const state = spectator.service.system_state(); - expect(state['sys-1'].disconnected).toEqual( - expect.arrayContaining(['Display_1', 'Audio_2']), - ); - }); -}); diff --git a/apps/stagehand/src/tests/broker-alerts.spec.ts b/apps/stagehand/src/tests/broker-alerts.spec.ts new file mode 100644 index 00000000000..088010c96b5 --- /dev/null +++ b/apps/stagehand/src/tests/broker-alerts.spec.ts @@ -0,0 +1,144 @@ +import { TriggerConditionOperator } from '@placeos/ts-client'; + +import { + BrokerValue, + evaluate, + parseBrokerMessage, + raisedAlerts, +} from '../app/dashboards/broker-alerts'; + +const value = ( + system: string, + device: string, + status: string, + v: unknown, + time = 100, +): BrokerValue => + parseBrokerMessage( + `placeos/org-1/state/_/bld/lvl/_/${system}/driver-1/${device}/1/${status}`, + JSON.stringify({ time, value: v }), + ); + +const definition = ( + comparisons: any[], + extra: Record = {}, +): any => ({ + id: 'alert-1', + name: 'Alert', + severity: 'high', + conditions: { comparisons }, + ...extra, +}); + +describe('broker alerts', () => { + it('reads an empty payload as a removed topic', () => { + expect(parseBrokerMessage('placeos/a/state/x', '')).toBeNull(); + expect( + parseBrokerMessage( + 'placeos/org-1/state/_/b/l/_/sys-1/d/Display/2/power', + '{"time":5,"value":true}', + ), + ).toEqual( + expect.objectContaining({ + time: 5, + value: true, + topic: expect.objectContaining({ + system_id: 'sys-1', + module_name: 'Display', + module_index: '2', + state_key: 'power', + }), + }), + ); + }); + + // the left of a comparison is the status value, the right what it is compared with + it('compares the status value with the expected one', () => { + expect(evaluate(30, TriggerConditionOperator.GT, 20)).toBe(true); + expect(evaluate(10, TriggerConditionOperator.GT, 20)).toBe(false); + expect(evaluate(10, TriggerConditionOperator.LT, 20)).toBe(true); + expect(evaluate('20', TriggerConditionOperator.GTE, 20)).toBe(true); + expect(evaluate('false', TriggerConditionOperator.EQ, false)).toBe( + true, + ); + expect(evaluate(true, TriggerConditionOperator.XOR, false)).toBe(true); + }); + + it('raises an alert per module whose value matches', () => { + const alerts = raisedAlerts( + [ + definition([ + { + left: 'connected', + operator: TriggerConditionOperator.EQ, + right: false, + }, + ]), + ], + [ + value('sys-1', 'Display', 'connected', false), + value('sys-1', 'Camera', 'connected', true), + value('sys-2', 'Display', 'connected', false), + ], + ); + expect(alerts.map((a) => a.id)).toEqual([ + 'alert-1+sys-1+Display_1', + 'alert-1+sys-2+Display_1', + ]); + expect(alerts[0]).toEqual( + expect.objectContaining({ timestamp: 100_000, status: 'open' }), + ); + }); + + it('reads sub-keys of a module status', () => { + const alerts = raisedAlerts( + [ + definition([ + { + left: { + mod: 'Display_1', + status: 'temperature', + keys: ['celsius'], + }, + operator: TriggerConditionOperator.GT, + right: 40, + }, + ]), + ], + [ + value('sys-1', 'Display', 'temperature', { celsius: 45 }), + value('sys-2', 'Display', 'temperature', { celsius: 30 }), + ], + ); + expect(alerts.map((a) => a.location)).toEqual(['sys-1']); + }); + + // every condition has to hold on the same system + it('groups the conditions of one alert by system', () => { + const alert = definition( + [ + { + left: { mod: 'Display_1', status: 'power' }, + operator: TriggerConditionOperator.EQ, + right: true, + }, + { + left: { mod: 'Projector_1', status: 'power' }, + operator: TriggerConditionOperator.EQ, + right: true, + }, + ], + { debounce_period: 60 }, + ); + const alerts = raisedAlerts( + [alert], + [ + value('sys-1', 'Display', 'power', true), + value('sys-2', 'Projector', 'power', true), + value('sys-3', 'Display', 'power', true), + value('sys-3', 'Projector', 'power', true), + ], + ); + expect(alerts.map((a) => a.location)).toEqual(['sys-3']); + }); +}); diff --git a/apps/stagehand/src/tests/dashboards.service.spec.ts b/apps/stagehand/src/tests/dashboards.service.spec.ts index 8a372c7230f..32dc570c5a7 100644 --- a/apps/stagehand/src/tests/dashboards.service.spec.ts +++ b/apps/stagehand/src/tests/dashboards.service.spec.ts @@ -321,54 +321,89 @@ describe('DashboardsService', () => { }); }); - // SH-HOT-02: the module list was cached on a system's first alert and never - // refreshed, so a module added afterwards could never raise an alert - describe('module existence check', () => { - const projector = { - id: 'mod-1', - custom_name: 'Projector', - driver: { class_name: 'Harness' }, - }; - const display = { - id: 'mod-2', - name: 'Display', - driver: { class_name: 'Harness' }, - }; - const exists = (device: string) => - (spectator.service as any)._moduleExistsOnSystem('sys-1', device); + // The broker is the source of truth: each topic's latest value raises or + // clears an alert, and nothing is checked again over REST + describe('broker messages', () => { + const topic = (system: string, device: string, index = 1) => + `placeos/org-1/state/_/bld/lvl/_/${system}/driver-1/${device}/${index}/connected`; + const send = (t: string, payload: unknown) => + (spectator.service as any)._handleBrokerMessage( + t, + Buffer.from( + payload === '' ? '' : JSON.stringify({ time: 1, value: payload }), + ), + ); + beforeEach(() => vi.useFakeTimers()); afterEach(() => vi.useRealTimers()); - it('refetches a stale module list before dropping an unknown device', async () => { - vi.useFakeTimers({ toFake: ['Date'] }); - vi.mocked(ts_client.queryModules) - .mockResolvedValueOnce({ data: [projector] } as any) - .mockResolvedValueOnce({ data: [projector, display] } as any); - - expect(await exists('Projector_1')).toBe(true); - vi.setSystemTime(Date.now() + 6_000); - - expect(await exists('Display_1')).toBe(true); - expect(ts_client.queryModules).toHaveBeenCalledTimes(2); + it('raises an alert from the latest value only', () => { + spectator.service.alerts_list.set([ + { + id: 'alert-1', + name: 'Disconnected', + conditions: { + comparisons: [ + { + left: 'connected', + operator: ts_client.TriggerConditionOperator.EQ, + right: false, + }, + ], + }, + } as any, + ]); + send(topic('sys-1', 'Display'), false); + send(topic('sys-2', 'Display'), false); + vi.advanceTimersByTime(100); + expect( + spectator.service.dashboard_alerts().map((a) => a.location), + ).toEqual(['sys-1', 'sys-2']); + + // the module reconnects: its earlier value no longer counts + send(topic('sys-1', 'Display'), true); + vi.advanceTimersByTime(100); + expect( + spectator.service.dashboard_alerts().map((a) => a.location), + ).toEqual(['sys-2']); + + expect(ts_client.queryModules).not.toHaveBeenCalled(); + expect(ts_client.systemModuleState).not.toHaveBeenCalled(); }); - it('refetches at most once per interval, and looks again when it may', async () => { - vi.useFakeTimers({ toFake: ['Date'] }); - vi.mocked(ts_client.queryModules).mockResolvedValue({ - data: [projector], - } as any); - const recheck = vi.spyOn(spectator.service as any, 'timeout'); - - expect(await exists('Projector_1')).toBe(true); - expect(await exists('Display_1')).toBe(false); - expect(await exists('Display_1')).toBe(false); + it('drops the alert of a topic the broker removes', () => { + spectator.service.alerts_list.set([ + { + id: 'alert-1', + name: 'Disconnected', + conditions: { + comparisons: [ + { + left: 'connected', + operator: ts_client.TriggerConditionOperator.EQ, + right: false, + }, + ], + }, + } as any, + ]); + send(topic('sys-1', 'Display', 2), false); + vi.advanceTimersByTime(100); + expect(spectator.service.dashboard_alerts()).toHaveLength(1); + + // source empties a retained topic when its module is deleted + send(topic('sys-1', 'Display', 2), ''); + vi.advanceTimersByTime(100); + expect(spectator.service.dashboard_alerts()).toEqual([]); + }); - expect(ts_client.queryModules).toHaveBeenCalledTimes(1); - expect(recheck).toHaveBeenCalledWith( - 'module_list_recheck', - expect.any(Function), - expect.any(Number), - ); + it('ignores a message it cannot read', () => { + expect(() => + (spectator.service as any)._handleBrokerMessage( + topic('sys-1', 'Display'), + Buffer.from('not json'), + ), + ).not.toThrow(); }); }); });