import { schedule } from "node-cron"; import { History, Patch } from "../db/models/index.js"; import { getLogger } from "../logger.js"; import { formatPowerStateForClient, getOrCreatePowerState, } from "./powerState.js"; import { getOrCreateWaterTank } from "./waterTank.js"; const logger = getLogger("History"); /** Every minute on the minute (system local time). */ const HISTORY_CRON = "* * * * *"; /** @type {import("node-cron").ScheduledTask | null} */ let cronTask = null; /** @type {Set<(payload: object) => void>} */ const powerStateListeners = new Set(); /** @type {Set<(payload: object) => void>} */ const waterTankListeners = new Set(); /** @type {Set<(payload: object) => void>} */ const patchListeners = new Set(); function numOrNull(v) { if (v == null || v === "") return null; const n = Number(v); return Number.isFinite(n) ? n : null; } /** * @param {import("mongoose").Document | Record} pow */ function snapshotPower(pow) { const { updatedAt: _u, ...rest } = formatPowerStateForClient(pow); return rest; } /** * @param {{ start: Date | string | number, end: Date | string | number }} range */ function parseTimeRange(range) { const { start, end } = range || {}; if (start == null || end == null) { throw new Error("start and end are required"); } const s = new Date(start); const e = new Date(end); if (Number.isNaN(s.getTime()) || Number.isNaN(e.getTime())) { throw new Error("Invalid start or end date"); } return { start: s, end: e }; } function queryForRange(range) { const { start, end } = parseTimeRange(range); return { sampledAt: { $gte: start, $lte: end }, }; } function emitPower(payload) { for (const fn of powerStateListeners) { try { fn(payload); } catch (e) { logger.error("powerState history listener error:", e); } } } function emitWaterTank(payload) { for (const fn of waterTankListeners) { try { fn(payload); } catch (e) { logger.error("waterTank history listener error:", e); } } } function emitPatch(payload) { for (const fn of patchListeners) { try { fn(payload); } catch (e) { logger.error("patch history listener error:", e); } } } export async function sampleHistoryOnce() { logger.trace("sampleHistoryOnce: fetching patch/tank/power state"); const [patchDocs, tank, pow] = await Promise.all([ Patch.find().select("patchId deviceId soilTemperature soilHumidity").lean(), getOrCreateWaterTank(), getOrCreatePowerState(), ]); logger.trace( "sampleHistoryOnce: loaded", patchDocs.length, "patches", "tank.level", numOrNull(tank.level), ); const sampledAt = new Date(); const patches = patchDocs.map((p) => ({ patchId: p.patchId, deviceId: p.deviceId, soilTemperature: numOrNull(p.soilTemperature), soilHumidity: numOrNull(p.soilHumidity), })); const powerSnapshot = snapshotPower(pow); const doc = await History.create({ sampledAt, waterTank: { level: numOrNull(tank.level) }, power: powerSnapshot, patches, }); const id = doc._id.toString(); logger.debug( "history sample created:", id, "sampledAt", doc.sampledAt.toISOString(), "patches", patches.length, "waterTank.level", doc.waterTank?.level ?? null, ); logger.trace( "history sample detail:", id, "batteryPercent", powerSnapshot?.battery?.percent ?? null, "solarVoltage", powerSnapshot?.solar?.voltage ?? null, "loadVoltage", powerSnapshot?.load?.voltage ?? null, ); emitPower({ historyId: id, sampledAt: doc.sampledAt, power: doc.power, }); emitWaterTank({ historyId: id, sampledAt: doc.sampledAt, waterTank: doc.waterTank, }); emitPatch({ historyId: id, sampledAt: doc.sampledAt, patches: doc.patches, }); return doc; } export function startHistorySampler() { if (cronTask != null) return; sampleHistoryOnce().catch((e) => { logger.error("Initial history sample failed:", e.message ?? e); }); cronTask = schedule(HISTORY_CRON, () => { sampleHistoryOnce().catch((e) => { logger.error("Periodic history sample failed:", e.message ?? e); }); }); } export function stopHistorySampler() { if (cronTask != null) { void cronTask.destroy(); cronTask = null; } } /** * @param {{ start: Date | string | number, end: Date | string | number }} range */ export async function getPowerStateHistory(range) { const filter = queryForRange(range); const rows = await History.find(filter) .select("sampledAt power") .sort({ sampledAt: 1 }) .lean(); return rows.map((r) => ({ historyId: r._id.toString(), sampledAt: r.sampledAt, power: r.power, })); } /** * @param {{ start: Date | string | number, end: Date | string | number }} range */ export async function getWaterTankHistory(range) { const filter = queryForRange(range); const rows = await History.find(filter) .select("sampledAt waterTank") .sort({ sampledAt: 1 }) .lean(); return rows.map((r) => ({ historyId: r._id.toString(), sampledAt: r.sampledAt, waterTank: r.waterTank, })); } /** * @param {string | { start: Date | string | number, end: Date | string | number }} patchIdOrRange * @param {{ start: Date | string | number, end: Date | string | number }} [range] * If `patchId` is omitted (call `getPatchHistory({ start, end })`), each row includes the full `patches` array for that sample. */ export async function getPatchHistory(patchIdOrRange, range) { let patchId; let r; if ( patchIdOrRange != null && typeof patchIdOrRange === "object" && !Array.isArray(patchIdOrRange) && ("start" in patchIdOrRange || "end" in patchIdOrRange) ) { r = patchIdOrRange; } else { patchId = patchIdOrRange == null || patchIdOrRange === "" ? undefined : String(patchIdOrRange); r = range; } const filter = queryForRange(r); const rows = await History.find(filter) .select("sampledAt patches") .sort({ sampledAt: 1 }) .lean(); if (!patchId) { return rows.map((row) => ({ historyId: row._id.toString(), sampledAt: row.sampledAt, patches: row.patches, })); } return rows .map((row) => { const p = (row.patches || []).find((x) => x.patchId === patchId); if (!p) return null; return { historyId: row._id.toString(), sampledAt: row.sampledAt, patchId: p.patchId, deviceId: p.deviceId?.toString?.() ?? p.deviceId, soilTemperature: p.soilTemperature, soilHumidity: p.soilHumidity, }; }) .filter(Boolean); } export function subscribePowerStateHistory(listener) { if (typeof listener !== "function") return; powerStateListeners.add(listener); } export function unsubscribePowerStateHistory(listener) { powerStateListeners.delete(listener); } export function subscribeWaterTankHistory(listener) { if (typeof listener !== "function") return; waterTankListeners.add(listener); } export function unsubscribeWaterTankHistory(listener) { waterTankListeners.delete(listener); } export function subscribePatchHistory(listener) { if (typeof listener !== "function") return; patchListeners.add(listener); } export function unsubscribePatchHistory(listener) { patchListeners.delete(listener); }