import { normalizeZ2mTopology, normalizeZhaTopology, type ZigbeeTopology } from './zigbee-topology'; import { normalizeZ2mBaseTopic } from './zigbee-topology-settings'; export type ZigbeeTopologyErrorCode = | 'permission' | 'unsupported' | 'timeout' | 'invalid_topic' | 'invalid_payload' | 'provider'; export type ZigbeeProviderState = { phase: 'idle' | 'loading' | 'ready' | 'error'; obtainedAt?: number; partial?: boolean; error?: ZigbeeTopologyErrorCode; }; export interface ZigbeeTopologyRuntimeSnapshot { revision: number; topologies: ZigbeeTopology[]; states: Record; } export interface ZigbeeTopologyHass { user?: { is_admin?: boolean }; language?: string; connection?: { subscribeMessage?: ( callback: (message: unknown) => void, message: { type: 'mqtt/subscribe'; topic: string }, ) => Promise<() => void>; }; callWS?: (message: { type: string }) => Promise; callService?: ( domain: string, service: string, data: Record, ) => Promise; } type Cache = ZigbeeTopologyRuntimeSnapshot & { listeners: Set<() => void>; inflight: Map>; }; const caches = new WeakMap(); function keyOf(hass: ZigbeeTopologyHass | null | undefined): object | null { const key = hass?.connection || hass; return key && (typeof key === 'object' || typeof key === 'function') ? key : null; } function cacheOf(hass: ZigbeeTopologyHass | null | undefined): Cache | null { const key = keyOf(hass); if (!key) return null; let cache = caches.get(key); if (!cache) { cache = { revision: 0, topologies: [], states: {}, listeners: new Set(), inflight: new Map() }; caches.set(key, cache); } return cache; } function notify(cache: Cache): void { cache.revision++; for (const listener of cache.listeners) listener(); } function state(cache: Cache, key: string, value: ZigbeeProviderState): void { cache.states = { ...cache.states, [key]: value }; notify(cache); } function store(cache: Cache, topology: ZigbeeTopology): void { cache.topologies = [ ...cache.topologies.filter((item) => !(item.provider === topology.provider && item.instanceId === topology.instanceId)), topology, ]; state(cache, topology.provider === 'zha' ? 'zha' : `z2m:${topology.instanceId}`, { phase: 'ready', obtainedAt: topology.obtainedAt, partial: topology.warnings.length > 0, }); } function errorCode(error: unknown): ZigbeeTopologyErrorCode { const code = (error as { code?: unknown } | null)?.code; if (code === 'permission' || code === 'unsupported' || code === 'timeout' || code === 'invalid_topic' || code === 'invalid_payload') return code; const message = String((error as { message?: unknown } | null)?.message || '').toLowerCase(); if (message.includes('unauthor') || message.includes('permission')) return 'permission'; if (message.includes('unknown_command') || message.includes('not found')) return 'unsupported'; return 'provider'; } function fail(code: ZigbeeTopologyErrorCode): Error & { code: ZigbeeTopologyErrorCode } { return Object.assign(new Error(code), { code }); } function run(cache: Cache, key: string, task: () => Promise): Promise { const existing = cache.inflight.get(key); if (existing) return existing; state(cache, key, { phase: 'loading' }); const promise = task().then((topology) => { if (!topology.nodes.length && topology.warnings.some((item) => item.code === 'invalid_payload')) { throw fail('invalid_payload'); } store(cache, topology); }).catch((error) => { state(cache, key, { phase: 'error', error: errorCode(error) }); }).finally(() => cache.inflight.delete(key)); cache.inflight.set(key, promise); return promise; } export function zigbeeTopologyRuntimeSnapshot( hass: ZigbeeTopologyHass | null | undefined, ): ZigbeeTopologyRuntimeSnapshot { const cache = cacheOf(hass); return cache ? { revision: cache.revision, topologies: cache.topologies, states: cache.states } : { revision: 0, topologies: [], states: {} }; } export function subscribeZigbeeTopology( hass: ZigbeeTopologyHass | null | undefined, listener: () => void, ): () => void { const cache = cacheOf(hass); if (!cache) return () => undefined; cache.listeners.add(listener); return () => cache.listeners.delete(listener); } function requireAdmin(hass: ZigbeeTopologyHass | null | undefined): void { if (hass?.user?.is_admin !== true) throw fail('permission'); } /** Read the cached ZHA graph. This must never call zha/topology/update. */ export function readZhaTopology(hass: ZigbeeTopologyHass): Promise { const cache = cacheOf(hass); if (!cache) return Promise.resolve(); return run(cache, 'zha', async () => { requireAdmin(hass); if (typeof hass?.callWS !== 'function') throw fail('unsupported'); return normalizeZhaTopology(await hass.callWS({ type: 'zha/devices' })); }); } function recordOf(value: unknown): Record | null { return value !== null && typeof value === 'object' ? value as Record : null; } function parseMessage(message: unknown): unknown { const record = recordOf(message); const raw = record?.payload ?? message; if (typeof raw !== 'string') return raw; try { return JSON.parse(raw); } catch { return null; } } function transactionOf(value: unknown): string | null { const record = recordOf(value); const transaction = record?.transaction ?? recordOf(record?.data)?.transaction; return typeof transaction === 'string' || typeof transaction === 'number' ? String(transaction) : null; } function randomTransaction(): string { const cryptoObj = globalThis.crypto; if (typeof cryptoObj?.randomUUID === 'function') return `houseplan-${cryptoObj.randomUUID()}`; return `houseplan-${Date.now().toString(36)}-${Math.random().toString(36).slice(2)}`; } async function withTimeout(promise: Promise, ms: number): Promise { let id: ReturnType | undefined; try { return await Promise.race([ promise, new Promise((_, reject) => { id = globalThis.setTimeout(() => reject(fail('timeout')), Math.max(1, ms)); }), ]); } finally { if (id !== undefined) globalThis.clearTimeout(id); } } /** Explicit, completion-aware Z2M raw network-map request through HA MQTT. */ export function refreshZ2mTopology( hass: ZigbeeTopologyHass, baseTopic: string, timeoutMs = 150_000, ): Promise { const cache = cacheOf(hass); if (!cache) return Promise.resolve(); const topic = normalizeZ2mBaseTopic(baseTopic); const cacheKey = `z2m:${topic || String(baseTopic)}`; return run(cache, cacheKey, async () => { requireAdmin(hass); if (!topic) throw fail('invalid_topic'); const connection = hass.connection; const subscribe = connection?.subscribeMessage; if (typeof subscribe !== 'function' || typeof hass?.callService !== 'function') throw fail('unsupported'); const transaction = randomTransaction(); const deadline = Date.now() + Math.max(1, timeoutMs); let responseActive = false; let infoResolve: (() => void) | null = null; let responseResolve: ((value: unknown) => void) | null = null; let responseReject: ((reason?: unknown) => void) | null = null; const info = new Promise((resolve) => { infoResolve = resolve; }); const response = new Promise((resolve, reject) => { responseResolve = resolve; responseReject = reject; }); const unsubscribers: Array<() => void> = []; try { const unsubInfo = await subscribe.call(connection, (message: unknown) => { if (recordOf(message)?.retain === true && parseMessage(message)) infoResolve?.(); }, { type: 'mqtt/subscribe', topic: `${topic}/bridge/info` }); if (typeof unsubInfo === 'function') unsubscribers.push(unsubInfo); const unsubResponse = await subscribe.call(connection, (message: unknown) => { if (recordOf(message)?.retain === true) return; const value = parseMessage(message); if (value === null) { if (responseActive) responseReject?.(fail('invalid_payload')); return; } if (value && transactionOf(value) === transaction) responseResolve?.(value); }, { type: 'mqtt/subscribe', topic: `${topic}/bridge/response/networkmap` }); if (typeof unsubResponse === 'function') unsubscribers.push(unsubResponse); await withTimeout(info, Math.min(4000, Math.max(1, deadline - Date.now()))); responseActive = true; await hass.callService('mqtt', 'publish', { topic: `${topic}/bridge/request/networkmap`, payload: JSON.stringify({ type: 'raw', routes: false, transaction }), qos: 0, retain: false, }); const value = await withTimeout(response, deadline - Date.now()); const status = recordOf(value)?.status; if (status && status !== 'ok') throw fail('provider'); return normalizeZ2mTopology(value, topic); } finally { for (const unsubscribe of unsubscribers) { try { unsubscribe(); } catch { /* cleanup is best effort */ } } } }); }