allotment/packages/server/socket/allotmentSocketApi.js

import { getLogger } from "../logger.js";
import {
  getDeviceState,
  getPatchState,
  getPowerStatePayload,
  getWaterTankState,
  getWateringSessionsState,
} from "../services/allotmentState.js";
import {
  startWateringSession,
  endWateringSession,
} from "../services/wateringSessions.js";
import { SubscriptionRegistry } from "./SubscriptionRegistry.js";
import { updatePatchForConfiguredDevice } from "../services/patches.js";
import {
  attachNodesToPatches,
  listNodes,
  normalizeSerial,
  setPatchNodes,
} from "../services/nodes.js";
import {
  getPatchHistory,
  getPowerStateHistory,
  getWaterTankHistory,
} from "../services/history.js";
import {
  createWaterSchedule,
  deleteWaterSchedule,
  listWaterSchedulesForConfiguredDevice,
  updateWaterSchedule,
} from "../services/waterSchedules.js";

const logger = getLogger("SocketApi");

function ackOk(data) {
  return { status: "ok", data };
}

function ackErr(message, code) {
  return { status: "error", error: message, ...(code ? { code } : {}) };
}

/**
 * @param {import("socket.io").Socket} socket
 * @param {object} deps
 * @param {import("./SocketClient.js").SocketClient} deps.socketClient
 * @param {SubscriptionRegistry} deps.subscriptionRegistry
 * @param {() => Promise<void>} deps.notifyWateringSubscribers
 * @param {() => Promise<void>} [deps.notifyWaterSchedulesSubscribers]
 * @param {(patch: object) => void} deps.emitPatchUpdated
 */
export function registerAllotmentSocketApi(socket, deps) {
  const {
    socketClient,
    subscriptionRegistry,
    notifyWateringSubscribers,
    notifyWaterSchedulesSubscribers,
    emitPatchUpdated,
  } = deps;

  const safeCb = (cb, payload) => {
    if (typeof cb === "function") cb(payload);
  };

  socket.on("getDevice", async (cb) => {
    try {
      const data = await getDeviceState();
      safeCb(cb, ackOk(data));
    } catch (e) {
      logger.error("getDevice", e);
      safeCb(cb, ackErr(e.message || "getDevice failed"));
    }
  });

  socket.on("getWaterTank", async (cb) => {
    try {
      const data = await getWaterTankState();
      safeCb(cb, ackOk(data));
    } catch (e) {
      logger.error("getWaterTank", e);
      safeCb(cb, ackErr(e.message || "getWaterTank failed"));
    }
  });

  socket.on("getPower", async (cb) => {
    try {
      const data = await getPowerStatePayload();
      safeCb(cb, ackOk(data));
    } catch (e) {
      logger.error("getPower", e);
      safeCb(cb, ackErr(e.message || "getPower failed"));
    }
  });

  socket.on("getWateringSessions", async (cb) => {
    try {
      const data = await getWateringSessionsState();
      safeCb(cb, ackOk(data));
    } catch (e) {
      logger.error("getWateringSessions", e);
      safeCb(cb, ackErr(e.message || "getWateringSessions failed"));
    }
  });

  socket.on("getWaterSchedules", async (cb) => {
    try {
      const data = await listWaterSchedulesForConfiguredDevice();
      safeCb(cb, ackOk(data));
    } catch (e) {
      logger.error("getWaterSchedules", e);
      safeCb(cb, ackErr(e.message || "getWaterSchedules failed"));
    }
  });

  socket.on("getNodes", async (cb) => {
    try {
      const nodes = await listNodes();
      safeCb(cb, ackOk({ nodes }));
    } catch (e) {
      logger.error("getNodes", e);
      safeCb(cb, ackErr(e.message || "getNodes failed"));
    }
  });

  socket.on("getPatch", async (payload, cb) => {
    try {
      const patchId = payload?.patchId;
      if (!patchId) {
        safeCb(cb, ackErr("patchId required"));
        return;
      }
      const patch = await getPatchState(patchId);
      if (!patch) {
        safeCb(cb, ackErr("Patch not found", "NOT_FOUND"));
        return;
      }
      safeCb(cb, ackOk({ patch }));
    } catch (e) {
      logger.error("getPatch", e);
      safeCb(cb, ackErr(e.message || "getPatch failed"));
    }
  });

  socket.on("getHistory", async (payload, cb) => {
    try {
      const kind = payload?.kind;
      const start = payload?.start;
      const end = payload?.end;
      const patchId = payload?.patchId;
      if (!kind || start == null || end == null) {
        safeCb(cb, ackErr("kind, start, and end are required"));
        return;
      }
      const range = { start, end };
      let data;
      if (kind === "power") {
        data = await getPowerStateHistory(range);
      } else if (kind === "waterTank") {
        data = await getWaterTankHistory(range);
      } else if (kind === "patch") {
        if (!patchId) {
          safeCb(cb, ackErr("patchId required for patch history"));
          return;
        }
        data = await getPatchHistory(patchId, range);
      } else {
        safeCb(cb, ackErr("Invalid kind"));
        return;
      }
      safeCb(cb, ackOk(data));
    } catch (e) {
      logger.error("getHistory", e);
      safeCb(cb, ackErr(e.message || "getHistory failed"));
    }
  });

  socket.on("editPatch", async (payload, cb) => {
    if (!socketClient.isWateringAllowed()) {
      safeCb(cb, ackErr("Sign in required to edit patches", "UNAUTHORIZED"));
      return;
    }
    try {
      const patchId = payload?.patchId;
      if (!patchId) {
        safeCb(cb, ackErr("patchId required"));
        return;
      }
      const nodes = payload?.nodes;
      if (nodes !== undefined) {
        if (!Array.isArray(nodes)) {
          safeCb(cb, ackErr("nodes must be a list of serials"));
          return;
        }
        nodes.forEach(normalizeSerial);
      }
      const updated = await updatePatchForConfiguredDevice(patchId, {
        name: payload?.name,
        description: payload?.description,
        icon: payload?.icon,
        rows: payload?.rows,
        columns: payload?.columns,
        lifecycle: payload?.lifecycle,
        plantedAt: payload?.plantedAt,
        sunlight: payload?.sunlight,
        soilPhMin: payload?.soilPhMin,
        soilPhMax: payload?.soilPhMax,
        waterNeeds: payload?.waterNeeds,
        images: payload?.images,
      });
      if (!updated) {
        safeCb(cb, ackErr("Patch not found", "NOT_FOUND"));
        return;
      }
      if (nodes !== undefined) {
        await setPatchNodes(updated.id, nodes);
      }
      const [patch] = await attachNodesToPatches([updated]);
      emitPatchUpdated(patch);
      safeCb(cb, ackOk({ patch }));
    } catch (e) {
      logger.error("editPatch", e);
      safeCb(cb, ackErr(e.message || "editPatch failed"));
    }
  });

  socket.on("subscribe", (msg, cb) => {
    const resource = msg?.resource;
    const id = msg?.id;
    const key = SubscriptionRegistry.channelKey(resource, id);
    if (!key) {
      safeCb(cb, ackErr("Invalid subscription resource or id"));
      return;
    }
    subscriptionRegistry.subscribe(socket.id, key);
    safeCb(cb, ackOk(null));
  });

  socket.on("unsubscribe", (msg, cb) => {
    const resource = msg?.resource;
    const id = msg?.id;
    const key = SubscriptionRegistry.channelKey(resource, id);
    if (!key) {
      safeCb(cb, ackErr("Invalid subscription resource or id"));
      return;
    }
    subscriptionRegistry.unsubscribe(socket.id, key);
    safeCb(cb, ackOk(null));
  });

  socket.on("startWateringSession", async (payload, cb) => {
    if (!socketClient.isWateringAllowed()) {
      safeCb(
        cb,
        ackErr("Sign in required to water plants", "UNAUTHORIZED"),
      );
      return;
    }
    try {
      const { patchIds } = payload || {};
      const sessionDoc = await startWateringSession({ patchIds });
      await notifyWateringSubscribers();
      safeCb(cb, {
        status: "ok",
        data: {
          ...sessionDoc,
          id: sessionDoc._id.toString(),
          deviceId: sessionDoc.deviceId?.toString(),
        },
      });
    } catch (e) {
      logger.error("startWateringSession", e);
      safeCb(cb, ackErr(e.message || "startWateringSession failed"));
    }
  });

  socket.on("endWateringSession", async (payload, cb) => {
    if (!socketClient.isWateringAllowed()) {
      safeCb(cb, ackErr("Sign in required", "UNAUTHORIZED"));
      return;
    }
    try {
      const { sessionId } = payload || {};
      const ended = await endWateringSession(sessionId);
      await notifyWateringSubscribers();
      safeCb(cb, {
        status: "ok",
        data: {
          ...ended,
          id: ended._id.toString(),
          deviceId: ended.deviceId?.toString(),
        },
      });
    } catch (e) {
      logger.error("endWateringSession", e);
      safeCb(cb, ackErr(e.message || "endWateringSession failed"));
    }
  });

  socket.on("createWaterSchedule", async (payload, cb) => {
    if (!socketClient.isWateringAllowed()) {
      safeCb(
        cb,
        ackErr("Sign in required to manage watering schedules", "UNAUTHORIZED"),
      );
      return;
    }
    try {
      const created = await createWaterSchedule({
        patchId: payload?.patchId,
        startAt: payload?.startAt,
        startAtType: payload?.startAtType,
        offset: payload?.offset,
        durationMinutes: payload?.durationMinutes,
        targetSoilHumidityEnabled: payload?.targetSoilHumidityEnabled,
        targetSoilHumidity: payload?.targetSoilHumidity,
        waterPower: payload?.waterPower,
      });
      if (notifyWaterSchedulesSubscribers) await notifyWaterSchedulesSubscribers();
      safeCb(cb, ackOk({ schedule: created }));
    } catch (e) {
      logger.error("createWaterSchedule", e);
      safeCb(cb, ackErr(e.message || "createWaterSchedule failed"));
    }
  });

  socket.on("updateWaterSchedule", async (payload, cb) => {
    if (!socketClient.isWateringAllowed()) {
      safeCb(
        cb,
        ackErr("Sign in required to manage watering schedules", "UNAUTHORIZED"),
      );
      return;
    }
    try {
      const updated = await updateWaterSchedule({
        scheduleId: payload?.scheduleId,
        patchId: payload?.patchId,
        startAt: payload?.startAt,
        startAtType: payload?.startAtType,
        offset: payload?.offset,
        durationMinutes: payload?.durationMinutes,
        targetSoilHumidityEnabled: payload?.targetSoilHumidityEnabled,
        targetSoilHumidity: payload?.targetSoilHumidity,
        waterPower: payload?.waterPower,
      });
      if (notifyWaterSchedulesSubscribers) await notifyWaterSchedulesSubscribers();
      safeCb(cb, ackOk({ schedule: updated }));
    } catch (e) {
      logger.error("updateWaterSchedule", e);
      safeCb(cb, ackErr(e.message || "updateWaterSchedule failed"));
    }
  });

  socket.on("deleteWaterSchedule", async (payload, cb) => {
    if (!socketClient.isWateringAllowed()) {
      safeCb(
        cb,
        ackErr("Sign in required to manage watering schedules", "UNAUTHORIZED"),
      );
      return;
    }
    try {
      const removed = await deleteWaterSchedule(payload?.scheduleId);
      if (notifyWaterSchedulesSubscribers) await notifyWaterSchedulesSubscribers();
      safeCb(cb, ackOk(removed));
    } catch (e) {
      logger.error("deleteWaterSchedule", e);
      safeCb(cb, ackErr(e.message || "deleteWaterSchedule failed"));
    }
  });
}