allotment/packages/server/services/history.js

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<string, unknown>} 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);
}