farmcontrol-ws/src/updates/updatemanager.js

import log4js from 'log4js';
import _ from 'lodash';
import { loadConfig } from '../config.js';
import { natsServer } from '../database/nats.js';
import { expandObjectIds } from '../database/utils.js';
import { getFilter, getObjectTypeModel } from '../database/filter.js';
import { formatTraceData } from '../utils.js';
const config = loadConfig();

// Setup logger
const logger = log4js.getLogger('Update Manager');
logger.level = config.server.logLevel;

const normalizeFilter = filter =>
  filter && typeof filter === 'object' && !Array.isArray(filter) ? filter : {};

const unwrapId = value => {
  if (
    value &&
    typeof value === 'object' &&
    !Array.isArray(value) &&
    value._id != null
  ) {
    return value._id;
  }
  return value;
};

const getFilterValue = (object, key) => {
  if (key.endsWith('._id')) {
    const refPath = key.slice(0, -4);
    const ref = _.get(object, refPath);
    if (ref && typeof ref === 'object' && ref._id) {
      return ref._id;
    }
    return ref;
  }

  return unwrapId(_.get(object, key));
};

const valuesMatch = (actual, expected) => {
  const left = unwrapId(actual);
  const right = unwrapId(expected);

  if (left == right) {
    return true;
  }

  if (left != null && right != null) {
    return String(left) === String(right);
  }

  return false;
};

const isOperatorObject = value => {
  if (!value || typeof value !== 'object' || Array.isArray(value)) {
    return false;
  }
  if (value instanceof Date || value._bsontype === 'ObjectId') {
    return false;
  }
  const keys = Object.keys(value);
  return keys.length > 0 && keys.every(key => key.startsWith('$'));
};

const compareValues = (actual, expected, operator) => {
  if (actual == null || expected == null) {
    return false;
  }
  if (operator === '$gt') return actual > expected;
  if (operator === '$gte') return actual >= expected;
  if (operator === '$lt') return actual < expected;
  if (operator === '$lte') return actual <= expected;
  return false;
};

const matchesOperators = (actual, operators) => {
  if (Object.prototype.hasOwnProperty.call(operators, '$eq')) {
    if (!valuesMatch(actual, operators.$eq)) return false;
  }
  if (Object.prototype.hasOwnProperty.call(operators, '$ne')) {
    if (valuesMatch(actual, operators.$ne)) return false;
  }
  if (Object.prototype.hasOwnProperty.call(operators, '$in')) {
    if (
      !Array.isArray(operators.$in) ||
      !operators.$in.some(value => valuesMatch(actual, value))
    ) {
      return false;
    }
  }
  if (Object.prototype.hasOwnProperty.call(operators, '$nin')) {
    if (
      Array.isArray(operators.$nin) &&
      operators.$nin.some(value => valuesMatch(actual, value))
    ) {
      return false;
    }
  }
  if (Object.prototype.hasOwnProperty.call(operators, '$regex')) {
    const regex = new RegExp(operators.$regex, operators.$options || '');
    if (!regex.test(String(actual ?? ''))) return false;
  }
  if (Object.prototype.hasOwnProperty.call(operators, '$not')) {
    if (matchesExpected(actual, operators.$not)) return false;
  }
  for (const operator of ['$gt', '$gte', '$lt', '$lte']) {
    if (
      Object.prototype.hasOwnProperty.call(operators, operator) &&
      !compareValues(actual, operators[operator], operator)
    ) {
      return false;
    }
  }
  return true;
};

const matchesExpected = (actual, expected) => {
  if (isOperatorObject(expected)) {
    return matchesOperators(actual, expected);
  }
  return valuesMatch(actual, expected);
};

const matchesFilter = (object, filter) => {
  if (!filter || Object.keys(filter).length === 0) {
    return true;
  }

  if (object == null) {
    return false;
  }

  const normalizedObject =
    typeof object === 'object' && !Array.isArray(object)
      ? object
      : { _id: object };

  if (Array.isArray(filter.$and)) {
    return filter.$and.every(clause => matchesFilter(normalizedObject, clause));
  }
  if (Array.isArray(filter.$or)) {
    return filter.$or.some(clause => matchesFilter(normalizedObject, clause));
  }

  for (const [key, expectedValue] of Object.entries(filter)) {
    if (key === '$and' || key === '$or') {
      continue;
    }
    if (
      !matchesExpected(getFilterValue(normalizedObject, key), expectedValue)
    ) {
      return false;
    }
  }

  return true;
};

const resolveObjectTypeFilter = async (objectType, filter = {}) => {
  const normalizedFilter = normalizeFilter(filter);
  if (Object.keys(normalizedFilter).length === 0) {
    return { normalizedFilter, processedFilter: normalizedFilter };
  }

  const model = await getObjectTypeModel(objectType);
  const processedFilter = await getFilter(
    normalizedFilter,
    Object.keys(normalizedFilter),
    true,
    model
  );
  return { normalizedFilter, processedFilter };
};

const stableStringify = value => {
  if (Array.isArray(value)) {
    return `[${value.map(stableStringify).join(',')}]`;
  }

  if (value && typeof value === 'object') {
    return `{${Object.keys(value)
      .sort()
      .map(key => `${JSON.stringify(key)}:${stableStringify(value[key])}`)
      .join(',')}}`;
  }

  return JSON.stringify(value);
};

const getSubscriptionOwner = (socketId, filter) =>
  `${socketId}:${stableStringify(normalizeFilter(filter))}`;

const getSubscriptionKey = (subject, owner) => `${subject}:${owner}`;

/**
 * UpdateManager handles tracking object updates and broadcasts update events via websockets.
 */
export class UpdateManager {
  constructor(socketClient) {
    this.socketClient = socketClient;
    this.subscriptions = new Set();
    this.objectUpdateSubscriptions = new Set();
  }

  getObjectUpdateSubscriptionKey(objectType, id) {
    return `${objectType}:${id}`;
  }

  matchesObjectTypeFilter(objectType, filter, value) {
    return matchesFilter(value, normalizeFilter(filter));
  }

  emitObjectTypeEvent(
    eventName,
    objectType,
    filter,
    value,
    matchFilter = filter
  ) {
    const normalizedFilter = normalizeFilter(filter);
    const matches = this.matchesObjectTypeFilter(
      objectType,
      matchFilter,
      value
    );

    if (!matches) {
      logger.trace(
        `Filtered ${eventName} event: ${formatTraceData({
          objectType,
          filter: normalizedFilter,
          value
        })}`
      );
      return;
    }

    this.socketClient.socket.emit(eventName, {
      object: value,
      objectType: objectType,
      filter: normalizedFilter
    });
  }

  async subscribeToObjectNew(objectType, filter = {}) {
    const { normalizedFilter, processedFilter } = await resolveObjectTypeFilter(
      objectType,
      filter
    );
    const subject = `${objectType}s.new`;
    const owner = getSubscriptionOwner(
      this.socketClient.socketId,
      normalizedFilter
    );

    await natsServer.subscribe(subject, owner, async (key, value) => {
      logger.trace(`Object new event: ${formatTraceData(value)}`);
      this.emitObjectTypeEvent(
        'objectNew',
        objectType,
        normalizedFilter,
        value,
        processedFilter
      );
    });

    this.subscriptions.add(getSubscriptionKey(subject, owner));
    return { success: true };
  }

  async subscribeToObjectDelete(objectType, filter = {}) {
    const { normalizedFilter, processedFilter } = await resolveObjectTypeFilter(
      objectType,
      filter
    );
    const subject = `${objectType}s.delete`;
    const owner = getSubscriptionOwner(
      this.socketClient.socketId,
      normalizedFilter
    );

    await natsServer.subscribe(subject, owner, async (key, value) => {
      logger.trace(`Object delete event: ${formatTraceData(value)}`);
      this.emitObjectTypeEvent(
        'objectDelete',
        objectType,
        normalizedFilter,
        value,
        processedFilter
      );
    });

    this.subscriptions.add(getSubscriptionKey(subject, owner));
    return { success: true };
  }

  async subscribeToObjectUpdate(id, objectType) {
    logger.debug('Subscribing to object update...', id, objectType);
    const subject = `${objectType}s.${id}.object`;
    const owner = this.socketClient.socketId;

    await natsServer.subscribe(subject, owner, (key, value) => {
      const expandedValue = expandObjectIds(value);
      logger.trace('Object update event:', id);
      this.socketClient.socket.emit('objectUpdate', {
        _id: id,
        objectType: objectType,
        object: { ...expandedValue }
      });
    });

    this.objectUpdateSubscriptions.add(
      this.getObjectUpdateSubscriptionKey(objectType, id)
    );
    this.subscriptions.add(getSubscriptionKey(subject, owner));
    return { success: true };
  }

  extractIdFromUpdateSubject(subject) {
    const parts = subject.split('.');
    if (parts.length < 3 || parts[parts.length - 1] !== 'object') {
      return null;
    }
    return parts[parts.length - 2];
  }

  async subscribeToAllObjectUpdates(objectType) {
    logger.debug('Subscribing to all object updates...', objectType);
    const subject = `${objectType}s.*.object`;
    const owner = this.socketClient.socketId;

    await natsServer.subscribe(subject, owner, (key, value) => {
      const id = this.extractIdFromUpdateSubject(key);
      if (!id) {
        logger.warn('Unable to extract id from update subject:', key);
        return;
      }

      if (
        this.objectUpdateSubscriptions.has(
          this.getObjectUpdateSubscriptionKey(objectType, id)
        )
      ) {
        return;
      }

      const expandedValue = expandObjectIds(value);
      logger.trace('All object update event:', id, objectType);
      this.socketClient.socket.emit('objectUpdate', {
        _id: id,
        objectType: objectType,
        object: { ...expandedValue }
      });
    });

    this.subscriptions.add(getSubscriptionKey(subject, owner));
    return { success: true };
  }

  async removeObjectNewListener(objectType, filter = {}) {
    const subject = `${objectType}s.new`;
    const owner = getSubscriptionOwner(this.socketClient.socketId, filter);

    await natsServer.removeSubscription(subject, owner);
    this.subscriptions.delete(getSubscriptionKey(subject, owner));
    return { success: true };
  }

  async removeObjectDeleteListener(objectType, filter = {}) {
    const subject = `${objectType}s.delete`;
    const owner = getSubscriptionOwner(this.socketClient.socketId, filter);

    await natsServer.removeSubscription(subject, owner);
    this.subscriptions.delete(getSubscriptionKey(subject, owner));
    return { success: true };
  }

  async removeObjectUpdateListener(id, objectType) {
    const subject = `${objectType}s.${id}.object`;
    const owner = this.socketClient.socketId;

    await natsServer.removeSubscription(subject, owner);
    this.objectUpdateSubscriptions.delete(
      this.getObjectUpdateSubscriptionKey(objectType, id)
    );
    this.subscriptions.delete(getSubscriptionKey(subject, owner));
    return { success: true };
  }

  async removeAllObjectUpdatesListener(objectType) {
    const subject = `${objectType}s.*.object`;
    const owner = this.socketClient.socketId;

    await natsServer.removeSubscription(subject, owner);
    this.subscriptions.delete(getSubscriptionKey(subject, owner));
    return { success: true };
  }

  async removeAllListeners() {
    logger.debug('Removing all update listeners...');
    const removePromises = Array.from(this.subscriptions).map(
      subscriptionKey => {
        const separatorIndex = subscriptionKey.indexOf(':');
        const subject = subscriptionKey.slice(0, separatorIndex);
        const owner = subscriptionKey.slice(separatorIndex + 1);
        return natsServer.removeSubscription(subject, owner);
      }
    );

    await Promise.all(removePromises);
    this.subscriptions.clear();
    this.objectUpdateSubscriptions.clear();
    logger.debug(`Removed ${removePromises.length} update listener(s)`);
    return { success: true };
  }
}