repo_description: Parse Server for Node.js / Express
repo_topics: baas backend file-storage graphql graphql-api graphql-relay graphql-server hacktoberfest mbaas mongodb nodejs notifications object-storage parse-platform parse-server postgres relay rest-api
code: import tv4 from 'tv4';
  import Parse from 'parse/node';
  import { Subscription } from './Subscription';
  import { Client } from './Client';
  import { ParseWebSocketServer } from './ParseWebSocketServer';
  // @ts-ignore
  import logger from '../logger';
  import RequestSchema from './RequestSchema';
  import { matchesQuery, queryHash } from './QueryTools';
  import { ParsePubSub } from './ParsePubSub';
  import SchemaController from '../Controllers/SchemaController';
  import _ from 'lodash';
  import { randomUUID } from 'crypto';
  import {
    runLiveQueryEventHandlers,
    getTrigger,
    runTrigger,
    resolveError,
    toJSONwithObjects,
  } from '../triggers';
  import { getAuthForSessionToken, Auth } from '../Auth';
  import { getCacheController, getDatabaseController } from '../Controllers';
  import Config from '../Config';
  import { LRUCache as LRU } from 'lru-cache';
  import UserRouter from '../Routers/UsersRouter';
  import DatabaseController from '../Controllers/DatabaseController';
  import { isDeepStrictEqual } from 'util';
  
  
  class ParseLiveQueryServer {
    server: any;
    config: any;
    clients: Map<string, any>;
    // className -> (queryHash -> subscription)
    subscriptions: Map<string, any>;
    parseWebSocketServer: any;
    keyPairs: any;
    // The subscriber we use to get object update from publisher
    subscriber: any;
    authCache: any;
    cacheController: any;
  
    constructor(server: any, config: any = {}, parseServerConfig: any = {}) {
      this.server = server;
      this.clients = new Map();
      this.subscriptions = new Map();
      this.config = config;
  
      config.appId = config.appId || Parse.applicationId;
      config.masterKey = config.masterKey || Parse.masterKey;
  
      // Store keys, convert obj to map
      const keyPairs = config.keyPairs || {};
      this.keyPairs = new Map();
      for (const key of Object.keys(keyPairs)) {
        this.keyPairs.set(key, keyPairs[key]);
      }
      logger.verbose('Support key pairs', this.keyPairs);
  
      // Initialize Parse
      Parse.Object.disableSingleInstance();
      const serverURL = config.serverURL || Parse.serverURL;
      Parse.serverURL = serverURL;
      Parse.initialize(config.appId, Parse.javaScriptKey, config.masterKey);
  
      // The cache controller is a proper cache controller
      // with access to User and Roles
      this.cacheController = getCacheController(parseServerConfig);
  
      config.cacheTimeout = config.cacheTimeout || 5 * 1000; // 5s
  
      // This auth cache stores the promises for each auth resolution.
      // The main benefit is to be able to reuse the same user / session token resolution.
      this.authCache = new LRU({
        max: 500, // 500 concurrent
        ttl: config.cacheTimeout,
      });
      // Initialize websocket server
      this.parseWebSocketServer = new ParseWebSocketServer(
        server,
        parseWebsocket => this._onConnect(parseWebsocket),
        config
      );
      this.subscriber = ParsePubSub.createSubscriber(config);
      if (!this.subscriber.connect) {
        this.connect();
      }
    }
  
    async connect() {
      if (this.subscriber.isOpen) {
        return;
      }
      if (typeof this.subscriber.connect === 'function') {
        await Promise.resolve(this.subscriber.connect());
      } else {
        this.subscriber.isOpen = true;
      }
      this._createSubscribers();
    }
  
    async shutdown() {
      if (this.subscriber.isOpen) {
        await Promise.all([
          ...[...this.clients.values()].map(client => client.parseWebSocket.ws.close()),
          this.parseWebSocketServer.close?.(),
          ...Array.from(this.subscriber.subscriptions?.keys() || []).map(key =>
            this.subscriber.unsubscribe(key)
          ),
          this.subscriber.close?.(),
        ]);
      }
      if (typeof this.subscriber.close === 'function') {
        try {
          await this.subscriber.close();
        } catch (err) {
          logger.error('PubSubAdapter error on shutdown', { error: err });
        }
      } else {
        this.subscriber.isOpen = false;
      }
    }
  
    _createSubscribers() {
      const messageRecieved = (channel, messageStr) => {
        logger.verbose('Subscribe message %j', messageStr);
        let message;
        try {
          message = JSON.parse(messageStr);
        } catch (e) {
          logger.error('unable to parse message', messageStr, e);
          return;
        }
        if (channel === Parse.applicationId + 'clearCache') {
          this._clearCachedRoles(message.userId);
          return;
        }
        this._inflateParseObject(message);
        if (channel === Parse.applicationId + 'afterSave') {
          this._onAfterSave(message);
        } else if (channel === Parse.applicationId + 'afterDelete') {
          this._onAfterDelete(message);
        } else {
          logger.error('Get message %s from unknown channel %j', message, channel);
        }
      };
      this.subscriber.on('message', (channel, messageStr) => messageRecieved(channel, messageStr));
      for (const field of ['afterSave', 'afterDelete', 'clearCache']) {
        const channel = `${Parse.applicationId}${field}`;
        this.subscriber.subscribe(channel, messageStr => messageRecieved(channel, messageStr));
      }
    }
  
    // Message is the JSON object from publisher. Message.currentParseObject is the ParseObject JSON after changes.
    // Message.originalParseObject is the original ParseObject JSON.
    _inflateParseObject(message: any): void {
      // Inflate merged object
      const currentParseObject = message.currentParseObject;
      UserRouter.removeHiddenProperties(currentParseObject);
      let className = currentParseObject.className;
      let parseObject = new Parse.Object(className);
      parseObject._finishFetch(currentParseObject);
      message.currentParseObject = parseObject;
      // Inflate original object
      const originalParseObject = message.originalParseObject;
      if (originalParseObject) {
        UserRouter.removeHiddenProperties(originalParseObject);
        className = originalParseObject.className;
        parseObject = new Parse.Object(className);
        parseObject._finishFetch(originalParseObject);
        message.originalParseObject = parseObject;
      }
    }
  
    // Message is the JSON object from publisher after inflated. Message.currentParseObject is the ParseObject after changes.
    // Message.originalParseObject is the original ParseObject.
    async _onAfterDelete(message: any): Promise<void> {
      logger.verbose(Parse.applicationId + 'afterDelete is triggered');
  
      let deletedParseObject = message.currentParseObject.toJSON();
      const classLevelPermissions = message.classLevelPermissions;
      const className = deletedParseObject.className;
      logger.verbose('ClassName: %j | ObjectId: %s', className, deletedParseObject.id);
      logger.verbose('Current client number : %d', this.clients.size);
  
      const classSubscriptions = this.subscriptions.get(className);
      if (typeof classSubscriptions === 'undefined') {
        logger.debug('Can not find subscriptions under this class ' + className);
        return;
      }
  
      for (const subscription of classSubscriptions.values()) {
        let isSubscriptionMatched;
        try {
          isSubscriptionMatched = this._matchesSubscription(deletedParseObject, subscription);
        } catch (e) {
          logger.error(`Failed matching subscription for class ${className}: ${e.message}`);
          continue;
        }
        if (!isSubscriptionMatched) {
          continue;
        }
        for (const [clientId, requestIds] of _.entries(subscription.clientRequestIds)) {
          const client = this.clients.get(clientId);
          if (typeof client === 'undefined') {
            continue;
          }
          requestIds.forEach(async requestId => {
            // Deep-clone shared object so each concurrent callback works on its own copy
            let localDeletedParseObject = JSON.parse(JSON.stringify(deletedParseObject));
            const acl = message.currentParseObject.getACL();
            // Check CLP
            const op = this._getCLPOperation(subscription.query);
            let res: any = {};
            try {
              const matchesCLP = await this._matchesCLP(
                classLevelPermissions,
                message.currentParseObject,
                client,
                requestId,
                op
              );
              if (matchesCLP === false) {
                return null;
              }
              const isMatched = await this._matchesACL(acl, client, requestId);
              if (!isMatched) {
                return null;
              }
              res = {
                event: 'delete',
                sessionToken: client.sessionToken,
                object: localDeletedParseObject,
                clients: this.clients.size,
                subscriptions: this.subscriptions.size,
                useMasterKey: client.hasMasterKey,
                installationId: client.installationId,
                sendEvent: true,
              };
              const trigger = getTrigger(className, 'afterEvent', Parse.applicationId);
              if (trigger) {
                const auth = await this.getAuthFromClient(client, requestId);
                if (auth && auth.user) {
                  res.user = auth.user;
                }
                if (res.object) {
                  res.object = Parse.Object.fromJSON(res.object);
                }
                await runTrigger(trigger, `afterEvent.${className}`, res, auth);
              }
              if (!res.sendEvent) {
                return;
              }
              if (res.object && typeof res.object.toJSON === 'function') {
                localDeletedParseObject = toJSONwithObjects(res.object, res.object.className || className);
              }
              res.object = localDeletedParseObject;
              await this._filterSensitiveData(
                classLevelPermissions,
                res,
                client,
                requestId,
                op,
                subscription.query
              );
              client.pushDelete(requestId, res.object);
            } catch (e) {
              const error = resolveError(e);
              Client.pushError(client.parseWebSocket, error.code, error.message, false, requestId);
              logger.error(
                `Failed running afterLiveQueryEvent on class ${className} for event ${res.event} with session ${res.sessionToken} with:\n Error: ` +
                  JSON.stringify(error)
              );
            }
          });
        }
      }
    }
  
    // Message is the JSON object from publisher after inflated. Message.currentParseObject is the ParseObject after changes.
    // Message.originalParseObject is the original ParseObject.
    async _onAfterSave(message: any): Promise<void> {
      logger.verbose(Parse.applicationId + 'afterSave is triggered');
  
      let originalParseObject = null;
      if (message.originalParseObject) {
        originalParseObject = message.originalParseObject.toJSON();
      }
      const classLevelPermissions = message.classLevelPermissions;
      let currentParseObject = message.currentParseObject.toJSON();
      const className = currentParseObject.className;
      logger.verbose('ClassName: %s | ObjectId: %s', className, currentParseObject.id);
      logger.verbose('Current client number : %d', this.clients.size);
  
      const classSubscriptions = this.subscriptions.get(className);
      if (typeof classSubscriptions === 'undefined') {
        logger.debug('Can not find subscriptions under this class ' + className);
        return;
      }
      for (const subscription of classSubscriptions.values()) {
        let isOriginalSubscriptionMatched;
        let isCurrentSubscriptionMatched;
        try {
          isOriginalSubscriptionMatched = this._matchesSubscription(
            originalParseObject,
            subscription
          );
          isCurrentSubscriptionMatched = this._matchesSubscription(
            currentParseObject,
            subscription
          );
        } catch (e) {
          logger.error(`Failed matching subscription for class ${className}: ${e.message}`);
          continue;
        }
        for (const [clientId, requestIds] of _.entries(subscription.clientRequestIds)) {
          const client = this.clients.get(clientId);
          if (typeof client === 'undefined') {
            continue;
          }
          requestIds.forEach(async requestId => {
            // Deep-clone shared objects so each concurrent callback works on its own copy.
            // Without cloning, _filterSensitiveData's in-place field deletion and afterEvent
            // trigger modifications corrupt the shared state across concurrent subscribers.
            let localCurrentParseObject = JSON.parse(JSON.stringify(currentParseObject));
            let localOriginalParseObject = originalParseObject
              ? JSON.parse(JSON.stringify(originalParseObject))
              : null;
            // Set orignal ParseObject ACL checking promise, if the object does not match
            // subscription, we do not need to check ACL
            let originalACLCheckingPromise;
            if (!isOriginalSubscriptionMatched) {
              originalACLCheckingPromise = Promise.resolve(false);
            } else {
              let originalACL;
              if (message.originalParseObject) {
                originalACL = message.originalParseObject.getACL();
              }
              originalACLCheckingPromise = this._matchesACL(originalACL, client, requestId);
            }
            // Set current ParseObject ACL checking promise, if the object does not match
            // subscription, we do not need to check ACL
            let currentACLCheckingPromise;
            let res: any = {};
            if (!isCurrentSubscriptionMatched) {
              currentACLCheckingPromise = Promise.resolve(false);
            } else {
              const currentACL = message.currentParseObject.getACL();
              currentACLCheckingPromise = this._matchesACL(currentACL, client, requestId);
            }
            try {
              const op = this._getCLPOperation(subscription.query);
              const matchesCLP = await this._matchesCLP(
                classLevelPermissions,
                message.currentParseObject,
                client,
                requestId,
                op
              );
              if (matchesCLP === false) {
                return;
              }
              const [isOriginalMatched, isCurrentMatched] = await Promise.all([
                originalACLCheckingPromise,
                currentACLCheckingPromise,
              ]);
              logger.verbose(
                'Original %j | Current %j | Match: %s, %s, %s, %s | Query: %s',
                localOriginalParseObject,
                localCurrentParseObject,
                isOriginalSubscriptionMatched,
                isCurrentSubscriptionMatched,
                isOriginalMatched,
                isCurrentMatched,
                subscription.hash
              );
              // Decide event type
              let type;
              if (isOriginalMatched && isCurrentMatched) {
                type = 'update';
              } else if (isOriginalMatched && !isCurrentMatched) {
                type = 'leave';
              } else if (!isOriginalMatched && isCurrentMatched) {
                if (localOriginalParseObject) {
                  type = 'enter';
                } else {
                  type = 'create';
                }
              } else {
                return null;
              }
              const watchFieldsChanged = this._checkWatchFields(client, requestId, message);
              if (!watchFieldsChanged && (type === 'update' || type === 'create')) {
                return;
              }
              res = {
                event: type,
                sessionToken: client.sessionToken,
                object: localCurrentParseObject,
                original: localOriginalParseObject,
                clients: this.clients.size,
                subscriptions: this.subscriptions.size,
                useMasterKey: client.hasMasterKey,
                installationId: client.installationId,
                sendEvent: true,
              };
              const trigger = getTrigger(className, 'afterEvent', Parse.applicationId);
              if (trigger) {
                if (res.object) {
                  res.object = Parse.Object.fromJSON(res.object);
                }
                if (res.original) {
                  res.original = Parse.Object.fromJSON(res.original);
                }
                const auth = await this.getAuthFromClient(client, requestId);
                if (auth && auth.user) {
                  res.user = auth.user;
                }
                await runTrigger(trigger, `afterEvent.${className}`, res, auth);
              }
              if (!res.sendEvent) {
                return;
              }
              if (res.object && typeof res.object.toJSON === 'function') {
                localCurrentParseObject = toJSONwithObjects(res.object, res.object.className || className);
              }
              if (res.original && typeof res.original.toJSON === 'function') {
                localOriginalParseObject = toJSONwithObjects(
                  res.original,
                  res.original.className || className
                );
              }
              res.object = localCurrentParseObject;
              res.original = localOriginalParseObject;
              await this._filterSensitiveData(
                classLevelPermissions,
                res,
                client,
                requestId,
                op,
                subscription.query
              );
              const functionName = 'push' + res.event.charAt(0).toUpperCase() + res.event.slice(1);
              if (client[functionName]) {
                client[functionName](requestId, res.object, res.original ?? null);
              }
            } catch (e) {
              const error = resolveError(e);
              Client.pushError(client.parseWebSocket, error.code, error.message, false, requestId);
              logger.error(
                `Failed running afterLiveQueryEvent on class ${className} for event ${res.event} with session ${res.sessionToken} with:\n Error: ` +
                  JSON.stringify(error)
              );
            }
          });
        }
      }
    }
  
    _onConnect(parseWebsocket: any): void {
      parseWebsocket.on('message', request => {
        if (typeof request === 'string') {
          try {
            request = JSON.parse(request);
          } catch (e) {
            logger.error('unable to parse request', request, e);
            return;
          }
        }
        logger.verbose('Request: %j', request);
  
        // Check whether this request is a valid request, return error directly if not
        if (
          !tv4.validate(request, RequestSchema['general']) ||
          !tv4.validate(request, RequestSchema[request.op])
        ) {
          Client.pushError(parseWebsocket, 1, tv4.error.message);
          logger.error('Connect message error %s', tv4.error.message);
          return;
        }
  
        switch (request.op) {
          case 'connect':
            this._handleConnect(parseWebsocket, request);
            break;
          case 'subscribe':
            this._handleSubscribe(parseWebsocket, request);
            break;
          case 'update':
            this._handleUpdateSubscription(parseWebsocket, request);
            break;
          case 'unsubscribe':
            this._handleUnsubscribe(parseWebsocket, request);
            break;
          default:
            Client.pushError(parseWebsocket, 3, 'Get unknown operation');
            logger.error('Get unknown operation', request.op);
        }
      });
  
      parseWebsocket.on('disconnect', () => {
        logger.info(`Client disconnect: ${parseWebsocket.clientId}`);
        const clientId = parseWebsocket.clientId;
        if (!this.clients.has(clientId)) {
          runLiveQueryEventHandlers({
            event: 'ws_disconnect_error',
            clients: this.clients.size,
            subscriptions: this.subscriptions.size,
            error: `Unable to find client ${clientId}`,
          });
          logger.error(`Can not find client ${clientId} on disconnect`);
          return;
        }
  
        // Delete client
        const client = this.clients.get(clientId);
        this.clients.delete(clientId);
  
        // Delete client from subscriptions
        for (const [requestId, subscriptionInfo] of _.entries(client.subscriptionInfos)) {
          const subscription = subscriptionInfo.subscription;
          subscription.deleteClientSubscription(clientId, requestId);
  
          // If there is no client which is subscribing this subscription, remove it from subscriptions
          const classSubscriptions = this.subscriptions.get(subscription.className);
          if (!subscription.hasSubscribingClient()) {
            classSubscriptions.delete(subscription.hash);
          }
          // If there is no subscriptions under this class, remove it from subscriptions
          if (classSubscriptions.size === 0) {
            this.subscriptions.delete(subscription.className);
          }
        }
  
        logger.verbose('Current clients %d', this.clients.size);
        logger.verbose('Current subscriptions %d', this.subscriptions.size);
        runLiveQueryEventHandlers({
          event: 'ws_disconnect',
          clients: this.clients.size,
          subscriptions: this.subscriptions.size,
          useMasterKey: client.hasMasterKey,
          installationId: client.installationId,
          sessionToken: client.sessionToken,
        });
      });
  
      runLiveQueryEventHandlers({
        event: 'ws_connect',
        clients: this.clients.size,
        subscriptions: this.subscriptions.size,
      });
    }
  
    _validateQueryConstraints(where: any): void {
      if (typeof where !== 'object' || where === null) {
        return;
      }
      for (const op of ['$or', '$and', '$nor']) {
        if (where[op] !== undefined && !Array.isArray(where[op])) {
          throw new Parse.Error(Parse.Error.INVALID_QUERY, `${op} must be an array`);
        }
        if (Array.isArray(where[op])) {
          where[op].forEach((subQuery: any) => {
            this._validateQueryConstraints(subQuery);
          });
        }
      }
      for (const key of Object.keys(where)) {
        const constraint = where[key];
        if (typeof constraint === 'object' && constraint !== null) {
          if (constraint.$regex !== undefined) {
            const regex = constraint.$regex;
            const isRegExpLike =
              regex !== null &&
              typeof regex === 'object' &&
              typeof regex.source === 'string' &&
              typeof regex.flags === 'string';
            if (typeof regex !== 'string' && !isRegExpLike) {
              throw new Parse.Error(
                Parse.Error.INVALID_QUERY,
                'Invalid regular expression: $regex must be a string or RegExp'
              );
            }
            const pattern = isRegExpLike ? regex.source : regex;
            const flags = isRegExpLike ? regex.flags : constraint.$options || '';
            try {
              new RegExp(pattern, flags);
            } catch (e) {
              throw new Parse.Error(
                Parse.Error.INVALID_QUERY,
                `Invalid regular expression: ${e.message}`
              );
            }
          }
        }
      }
    }
  
    _matchesSubscription(parseObject: any, subscription: any): boolean {
      // Object is undefined or null, not match
      if (!parseObject) {
        return false;
      }
      return matchesQuery(structuredClone(parseObject), subscription.query);
    }
  
    async _clearCachedRoles(userId: string) {
      try {
        const validTokens = await new Parse.Query(Parse.Session)
          .equalTo('user', Parse.User.createWithoutData(userId))
          .find({ useMasterKey: true });
        await Promise.all(
          validTokens.map(async token => {
            const sessionToken = token.get('sessionToken');
            const authPromise = this.authCache.get(sessionToken);
            if (!authPromise) {
              return;
            }
            const [auth1, auth2] = await Promise.all([
              authPromise,
              getAuthForSessionToken({ cacheController: this.cacheController, sessionToken }),
            ]);
            auth1.auth?.clearRoleCache(sessionToken);
            auth2.auth?.clearRoleCache(sessionToken);
            this.authCache.delete(sessionToken);
          })
        );
      } catch (e) {
        logger.verbose(`Could not clear role cache. ${e}`);
      }
    }
  
    getAuthForSessionToken(sessionToken?: string): Promise<{ auth?: Auth, userId?: string }> {
      if (!sessionToken) {
        return Promise.resolve({});
      }
      const fromCache = this.authCache.get(sessionToken);
      if (fromCache) {
        return fromCache;
      }
      const authPromise = getAuthForSessionToken({
        cacheController: this.cacheController,
        sessionToken: sessionToken,
      })
        .then(auth => {
          return { auth, userId: auth && auth.user && auth.user.id };
        })
        .catch(error => {
          // There was an error with the session token
          const result: any = {};
          if (error && error.code === Parse.Error.INVALID_SESSION_TOKEN) {
            result.error = error;
            this.authCache.set(sessionToken, Promise.resolve(result), this.config.cacheTimeout);
          } else {
            this.authCache.delete(sessionToken);
          }
          return result;
        });
      this.authCache.set(sessionToken, authPromise);
      return authPromise;
    }
  
    async _matchesCLP(
      classLevelPermissions?: any,
      object?: any,
      client?: any,
      requestId?: number,
      op?: string
    ): Promise<any> {
      const subscriptionInfo = client.getSubscriptionInfo(requestId);
      const aclGroup = ['*'];
      let userId;
      if (typeof subscriptionInfo !== 'undefined') {
        const result = await this.getAuthForSessionToken(subscriptionInfo.sessionToken);
        userId = result.userId;
        if (userId) {
          aclGroup.push(userId);
        }
      }
      await SchemaController.validatePermission(
        classLevelPermissions,
        object.className,
        aclGroup,
        op
      );
      // Enforce pointer permissions that validatePermission defers.
      // Returns false to silently skip the event (like ACL), rather than
      // throwing which would push errors to the client and log noise.
      if (!client.hasMasterKey && classLevelPermissions) {
        const permissionField =
          ['get', 'find', 'count'].indexOf(op) > -1 ? 'readUserFields' : 'writeUserFields';
        const pointerFields = [];
        if (classLevelPermissions[op]?.pointerFields) {
          pointerFields.push(...classLevelPermissions[op].pointerFields);
        }
        if (Array.isArray(classLevelPermissions[permissionField])) {
          for (const field of classLevelPermissions[permissionField]) {
            if (!pointerFields.includes(field)) {
              pointerFields.push(field);
            }
          }
        }
        if (pointerFields.length > 0) {
          // If public or user-specific permission already grants access, skip pointer check
          if (
            !SchemaController.testPermissions(classLevelPermissions, aclGroup, op)
          ) {
            if (!userId) {
              return false;
            }
            // Check if any pointer field points to the current user
            const hasAccess = pointerFields.some(field => {
              const value =
                typeof object.get === 'function' ? object.get(field) : object[field];
              if (!value) {
                return false;
              }
              // Handle Parse.Object pointer (has .id)
              if (value.id) {
                return value.id === userId;
              }
              // Handle raw pointer JSON (has .objectId)
              if (value.objectId) {
                return value.objectId === userId;
              }
              // Handle array of pointers
              if (Array.isArray(value)) {
                return value.some(item => {
                  if (item.id) {
                    return item.id === userId;
                  }
                  if (item.objectId) {
                    return item.objectId === userId;
                  }
                  return false;
                });
              }
              return false;
            });
            if (!hasAccess) {
              return false;
            }
          }
        }
      }
    }
  
    async _filterSensitiveData(
      classLevelPermissions?: any,
      res?: any,
      client?: any,
      requestId?: number,
      op?: string,
      query?: any
    ) {
      const subscriptionInfo = client.getSubscriptionInfo(requestId);
      const aclGroup = ['*'];
      let clientAuth;
      if (typeof subscriptionInfo !== 'undefined') {
        const { userId, auth } = await this.getAuthForSessionToken(subscriptionInfo.sessionToken);
        if (userId) {
          aclGroup.push(userId);
        }
        clientAuth = auth;
      }
      const filter = obj => {
        if (!obj) {
          return;
        }
        let protectedFields = classLevelPermissions?.protectedFields || [];
        if (client.hasMasterKey) {
          protectedFields = [];
        } else if (!Array.isArray(protectedFields)) {
          protectedFields = getDatabaseController(this.config).addProtectedFields(
            classLevelPermissions,
            res.object.className,
            query,
            aclGroup,
            clientAuth
          );
        }
        return DatabaseController.filterSensitiveData(
          client.hasMasterKey,
          false,
          aclGroup,
          clientAuth,
          op,
          classLevelPermissions,
          res.object.className,
          protectedFields,
          obj,
          this.config.protectedFieldsOwnerExempt
        );
      };
      res.object = filter(res.object);
      res.original = filter(res.original);
    }
  
    _getCLPOperation(query: any) {
      return typeof query === 'object' &&
        Object.keys(query).length == 1 &&
        typeof query.objectId === 'string'
        ? 'get'
        : 'find';
    }
  
    async _verifyACL(acl: any, token: string) {
      if (!token) {
        return false;
      }
  
      const { auth, userId } = await this.getAuthForSessionToken(token);
  
      // Getting the session token failed
      // This means that no additional auth is available
      // At this point, just bail out as no additional visibility can be inferred.
      if (!auth || !userId) {
        return false;
      }
      const isSubscriptionSessionTokenMatched = acl.getReadAccess(userId);
      if (isSubscriptionSessionTokenMatched) {
        return true;
      }
  
      // Check if the user has any roles that match the ACL
      return Promise.resolve()
        .then(async () => {
          // Resolve false right away if the acl doesn't have any roles
          const acl_has_roles = Object.keys(acl.permissionsById).some(key => key.startsWith('role:'));
          if (!acl_has_roles) {
            return false;
          }
          const roleNames = await auth.getUserRoles();
          // Finally, see if any of the user's roles allow them read access
          for (const role of roleNames) {
            // We use getReadAccess as `role` is in the form `role:roleName`
            if (acl.getReadAccess(role)) {
              return true;
            }
          }
          return false;
        })
        .catch(() => {
          return false;
        });
    }
  
    async getAuthFromClient(client: any, requestId: number, sessionToken?: string) {
      const getSessionFromClient = () => {
        const subscriptionInfo = client.getSubscriptionInfo(requestId);
        if (typeof subscriptionInfo === 'undefined') {
          return client.sessionToken;
        }
        return subscriptionInfo.sessionToken || client.sessionToken;
      };
      if (!sessionToken) {
        sessionToken = getSessionFromClient();
      }
      if (!sessionToken) {
        return;
      }
      const { auth } = await this.getAuthForSessionToken(sessionToken);
      return auth;
    }
  
    _checkWatchFields(client: any, requestId: any, message: any) {
      const subscriptionInfo = client.getSubscriptionInfo(requestId);
      const watch = subscriptionInfo?.watch;
      if (!watch) {
        return true;
      }
      const object = message.currentParseObject;
      const original = message.originalParseObject;
      return watch.some(field => !isDeepStrictEqual(object.get(field), original?.get(field)));
    }
  
    async _matchesACL(acl: any, client: any, requestId: number): Promise<boolean> {
      // Return true directly if ACL isn't present, ACL is public read, or client has master key
      if (!acl || acl.getPublicReadAccess() || client.hasMasterKey) {
        return true;
      }
      // Check subscription sessionToken matches ACL first
      const subscriptionInfo = client.getSubscriptionInfo(requestId);
      if (typeof subscriptionInfo === 'undefined') {
        return false;
      }
  
      const subscriptionToken = subscriptionInfo.sessionToken;
      const clientSessionToken = client.sessionToken;
  
      if (await this._verifyACL(acl, subscriptionToken)) {
        return true;
      }
  
      if (await this._verifyACL(acl, clientSessionToken)) {
        return true;
      }
  
      return false;
    }
  
    async _handleConnect(parseWebsocket: any, request: any): Promise<any> {
      if (!this._validateKeys(request, this.keyPairs)) {
        Client.pushError(parseWebsocket, 4, 'Key in request is not valid');
        logger.error('Key in request is not valid');
        return;
      }
      const hasMasterKey = this._hasMasterKey(request, this.keyPairs);
      const clientId = randomUUID();
      const client = new Client(
        clientId,
        parseWebsocket,
        hasMasterKey,
        request.sessionToken,
        request.installationId
      );
      try {
        const req = {
          client,
          event: 'connect',
          clients: this.clients.size,
          subscriptions: this.subscriptions.size,
          sessionToken: request.sessionToken,
          useMasterKey: client.hasMasterKey,
          installationId: request.installationId,
          user: undefined,
        };
        const trigger = getTrigger('@Connect', 'beforeConnect', Parse.applicationId);
        if (trigger) {
          const auth = await this.getAuthFromClient(client, request.requestId, req.sessionToken);
          if (auth && auth.user) {
            req.user = auth.user;
          }
          await runTrigger(trigger, `beforeConnect.@Connect`, req, auth);
        }
        parseWebsocket.clientId = clientId;
        this.clients.set(parseWebsocket.clientId, client);
        logger.info(`Create new client: ${parseWebsocket.clientId}`);
        client.pushConnect();
        runLiveQueryEventHandlers(req);
      } catch (e) {
        const error = resolveError(e);
        Client.pushError(parseWebsocket, error.code, error.message, false);
        logger.error(
          `Failed running beforeConnect for session ${request.sessionToken} with:\n Error: ` +
            JSON.stringify(error)
        );
      }
    }
  
    _hasMasterKey(request: any, validKeyPairs: any): boolean {
      if (!validKeyPairs || validKeyPairs.size == 0 || !validKeyPairs.has('masterKey')) {
        return false;
      }
      if (!request || !Object.prototype.hasOwnProperty.call(request, 'masterKey')) {
        return false;
      }
      return request.masterKey === validKeyPairs.get('masterKey');
    }
  
    _validateKeys(request: any, validKeyPairs: any): boolean {
      if (!validKeyPairs || validKeyPairs.size == 0) {
        return true;
      }
      let isValid = false;
      for (const [key, secret] of validKeyPairs) {
        if (!request[key] || request[key] !== secret) {
          continue;
        }
        isValid = true;
        break;
      }
      return isValid;
    }
  
    async _handleSubscribe(parseWebsocket: any, request: any): Promise<any> {
      // If we can not find this client, return error to client
      if (!Object.prototype.hasOwnProperty.call(parseWebsocket, 'clientId')) {
        Client.pushError(
          parseWebsocket,
          2,
          'Can not find this client, make sure you connect to server before subscribing'
        );
        logger.error('Can not find this client, make sure you connect to server before subscribing');
        return;
      }
      const client = this.clients.get(parseWebsocket.clientId);
      const className = request.query.className;
      let authCalled = false;
      try {
        const trigger = getTrigger(className, 'beforeSubscribe', Parse.applicationId);
        if (trigger) {
          const auth = await this.getAuthFromClient(client, request.requestId, request.sessionToken);
          authCalled = true;
          if (auth && auth.user) {
            request.user = auth.user;
          }
  
          const parseQuery = new Parse.Query(className);
          parseQuery.withJSON(request.query);
          request.query = parseQuery;
          await runTrigger(trigger, `beforeSubscribe.${className}`, request, auth);
  
          const query = request.query.toJSON();
          request.query = query;
        }
  
        if (className === '_Session') {
          if (!authCalled) {
            const auth = await this.getAuthFromClient(
              client,
              request.requestId,
              request.sessionToken
            );
            if (auth && auth.user) {
              request.user = auth.user;
            }
          }
          if (request.user) {
            request.query.where.user = request.user.toPointer();
          } else if (!request.master) {
            Client.pushError(
              parseWebsocket,
              Parse.Error.INVALID_SESSION_TOKEN,
              'Invalid session token',
              false,
              request.requestId
            );
            return;
          }
        }
        // Validate query condition depth
        const appConfig = Config.get(this.config.appId);
        if (!client.hasMasterKey) {
          const rc = appConfig.requestComplexity;
          if (rc && rc.queryDepth !== -1) {
            const maxDepth = rc.queryDepth;
            const checkDepth = (where: any, depth: number) => {
              if (depth > maxDepth) {
                throw new Parse.Error(
                  Parse.Error.INVALID_QUERY,
                  `Query condition nesting depth exceeds maximum allowed depth of ${maxDepth}`
                );
              }
              if (typeof where !== 'object' || where === null) {
                return;
              }
              for (const op of ['$or', '$and', '$nor']) {
                if (where[op] !== undefined && !Array.isArray(where[op])) {
                  throw new Parse.Error(Parse.Error.INVALID_QUERY, `${op} must be an array`);
                }
                if (Array.isArray(where[op])) {
                  for (const subQuery of where[op]) {
                    checkDepth(subQuery, depth + 1);
                  }
                }
              }
            };
            checkDepth(request.query.where, 0);
          }
        }
  
        // Validate allowRegex
        if (!client.hasMasterKey) {
          const rc = appConfig.requestComplexity;
          if (rc && rc.allowRegex === false) {
            const checkRegex = (where: any) => {
              if (typeof where !== 'object' || where === null) {
                return;
              }
              for (const key of Object.keys(where)) {
                const constraint = where[key];
                if (typeof constraint === 'object' && constraint !== null && constraint.$regex !== undefined) {
                  throw new Parse.Error(Parse.Error.INVALID_QUERY, '$regex operator is not allowed');
                }
              }
              for (const op of ['$or', '$and', '$nor']) {
                if (Array.isArray(where[op])) {
                  for (const subQuery of where[op]) {
                    checkRegex(subQuery);
                  }
                }
              }
            };
            checkRegex(request.query.where);
          }
        }
  
        // Check CLP for subscribe operation
        const schemaController = await appConfig.database.loadSchema();
        const classLevelPermissions = schemaController.getClassLevelPermissions(className);
        const op = this._getCLPOperation(request.query);
        const aclGroup = ['*'];
        if (!authCalled) {
          const auth = await this.getAuthFromClient(
            client,
            request.requestId,
            request.sessionToken
          );
          authCalled = true;
          if (auth && auth.user) {
            request.user = auth.user;
            aclGroup.push(auth.user.id);
          }
        } else if (request.user) {
          aclGroup.push(request.user.id);
        }
        await SchemaController.validatePermission(
          classLevelPermissions,
          className,
          aclGroup,
          op
        );
  
        // Check protected fields in WHERE clause and WATCH parameter
        if (!client.hasMasterKey) {
          const auth = request.user ? { user: request.user, userRoles: [] } : {};
          const protectedFields =
            appConfig.database.addProtectedFields(
              classLevelPermissions,
              className,
              request.query.where,
              aclGroup,
              auth
            ) || [];
          if (protectedFields.length > 0 && request.query.where) {
            const checkWhere = (where: any) => {
              if (typeof where !== 'object' || where === null) {
                return;
              }
              for (const whereKey of Object.keys(where)) {
                const rootField = whereKey.split('.')[0];
                if (protectedFields.includes(whereKey) || protectedFields.includes(rootField)) {
                  throw new Parse.Error(
                    Parse.Error.OPERATION_FORBIDDEN,
                    'Permission denied'
                  );
                }
              }
              for (const op of ['$or', '$and', '$nor']) {
                if (where[op] !== undefined && !Array.isArray(where[op])) {
                  throw new Parse.Error(Parse.Error.INVALID_QUERY, `${op} must be an array`);
                }
                if (Array.isArray(where[op])) {
                  where[op].forEach((subQuery: any) => checkWhere(subQuery));
                }
              }
            };
            checkWhere(request.query.where);
          }
          if (protectedFields.length > 0 && Array.isArray(request.query.watch)) {
            for (const watchField of request.query.watch) {
              const rootField = watchField.split('.')[0];
              if (protectedFields.includes(watchField) || protectedFields.includes(rootField)) {
                throw new Parse.Error(
                  Parse.Error.OPERATION_FORBIDDEN,
                  'Permission denied'
                );
              }
            }
          }
        }
  
        // Validate regex patterns in the subscription query
        this._validateQueryConstraints(request.query.where);
  
        // Get subscription from subscriptions, create one if necessary
        const subscriptionHash = queryHash(request.query);
        // Add className to subscriptions if necessary
  
        if (!this.subscriptions.has(className)) {
          this.subscriptions.set(className, new Map());
        }
        const classSubscriptions = this.subscriptions.get(className);
        let subscription;
        if (classSubscriptions.has(subscriptionHash)) {
          subscription = classSubscriptions.get(subscriptionHash);
        } else {
          subscription = new Subscription(className, request.query.where, subscriptionHash);
          classSubscriptions.set(subscriptionHash, subscription);
        }
  
        // Add subscriptionInfo to client
        const subscriptionInfo: any = {
          subscription: subscription,
        };
        // Add selected fields, sessionToken and installationId for this subscription if necessary
        if (request.query.keys) {
          subscriptionInfo.keys = Array.isArray(request.query.keys)
            ? request.query.keys
            : request.query.keys.split(',');
        }
        if (request.query.watch) {
          subscriptionInfo.watch = request.query.watch;
        }
        if (request.sessionToken) {
          subscriptionInfo.sessionToken = request.sessionToken;
        }
        client.addSubscriptionInfo(request.requestId, subscriptionInfo);
  
        // Add clientId to subscription
        subscription.addClientSubscription(parseWebsocket.clientId, request.requestId);
  
        client.pushSubscribe(request.requestId);
  
        logger.verbose(
          `Create client ${parseWebsocket.clientId} new subscription: ${request.requestId}`
        );
        logger.verbose('Current client number: %d', this.clients.size);
        runLiveQueryEventHandlers({
          client,
          event: 'subscribe',
          clients: this.clients.size,
          subscriptions: this.subscriptions.size,
          sessionToken: request.sessionToken,
          useMasterKey: client.hasMasterKey,
          installationId: client.installationId,
        });
      } catch (e) {
        const error = resolveError(e);
        Client.pushError(parseWebsocket, error.code, error.message, false, request.requestId);
        logger.error(
          `Failed running beforeSubscribe on ${className} for session ${request.sessionToken} with:\n Error: ` +
            JSON.stringify(error)
        );
      }
    }
  
    _handleUpdateSubscription(parseWebsocket: any, request: any): any {
      this._handleUnsubscribe(parseWebsocket, request, false);
      this._handleSubscribe(parseWebsocket, request);
    }
  
    _handleUnsubscribe(parseWebsocket: any, request: any, notifyClient: boolean = true): any {
      // If we can not find this client, return error to client
      if (!Object.prototype.hasOwnProperty.call(parseWebsocket, 'clientId')) {
        Client.pushError(
          parseWebsocket,
          2,
          'Can not find this client, make sure you connect to server before unsubscribing'
        );
        logger.error(
          'Can not find this client, make sure you connect to server before unsubscribing'
        );
        return;
      }
      const requestId = request.requestId;
      const client = this.clients.get(parseWebsocket.clientId);
      if (typeof client === 'undefined') {
        Client.pushError(
          parseWebsocket,
          2,
          'Cannot find client with clientId ' +
            parseWebsocket.clientId +
            '. Make sure you connect to live query server before unsubscribing.'
        );
        logger.error('Can not find this client ' + parseWebsocket.clientId);
        return;
      }
  
      const subscriptionInfo = client.getSubscriptionInfo(requestId);
      if (typeof subscriptionInfo === 'undefined') {
        Client.pushError(
          parseWebsocket,
          2,
          'Cannot find subscription with clientId ' +
            parseWebsocket.clientId +
            ' subscriptionId ' +
            requestId +
            '. Make sure you subscribe to live query server before unsubscribing.'
        );
        logger.error(
          'Can not find subscription with clientId ' +
            parseWebsocket.clientId +
            ' subscriptionId ' +
            requestId
        );
        return;
      }
  
      // Remove subscription from client
      client.deleteSubscriptionInfo(requestId);
      // Remove client from subscription
      const subscription = subscriptionInfo.subscription;
      const className = subscription.className;
      subscription.deleteClientSubscription(parseWebsocket.clientId, requestId);
      // If there is no client which is subscribing this subscription, remove it from subscriptions
      const classSubscriptions = this.subscriptions.get(className);
      if (!subscription.hasSubscribingClient()) {
        classSubscriptions.delete(subscription.hash);
      }
      // If there is no subscriptions under this class, remove it from subscriptions
      if (classSubscriptions.size === 0) {
        this.subscriptions.delete(className);
      }
      runLiveQueryEventHandlers({
        client,
        event: 'unsubscribe',
        clients: this.clients.size,
        subscriptions: this.subscriptions.size,
        sessionToken: subscriptionInfo.sessionToken,
        useMasterKey: client.hasMasterKey,
        installationId: client.installationId,
      });
  
      if (!notifyClient) {
        return;
      }
  
      client.pushUnsubscribe(request.requestId);
  
      logger.verbose(
        `Delete client: ${parseWebsocket.clientId} | subscription: ${request.requestId}`
      );
    }
  }
  
  export { ParseLiveQueryServer };
  