mirror of
https://github.com/parse-community/parse-server
synced 2026-08-09 13:03:18 +00:00
1381 lines
50 KiB
TypeScript
1381 lines
50 KiB
TypeScript
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;
|
|
}
|
|
// A `leave` or `enter` transition can be caused either by the object's
|
|
// query match changing (the subscriber keeps read access) or by the
|
|
// subscriber's ACL read access being revoked or granted in the same save.
|
|
// In the access-change case the subscriber is not authorized to read the
|
|
// object state that triggered the transition, so that state must not be
|
|
// sent over the channel. (CLP read denial is handled earlier by
|
|
// `_matchesCLP`, which skips the event entirely.)
|
|
if (type === 'leave') {
|
|
// The post-update object is readable on a query-mismatch leave but not
|
|
// on an ACL-loss leave. Only send the post-update body when the
|
|
// subscriber can still read the current object; otherwise fall back to
|
|
// the last authorized (original) state, which still carries the objectId.
|
|
const currentReadable = isCurrentSubscriptionMatched
|
|
? false
|
|
: await this._matchesACL(message.currentParseObject.getACL(), client, requestId);
|
|
if (!currentReadable) {
|
|
localCurrentParseObject = JSON.parse(JSON.stringify(localOriginalParseObject));
|
|
}
|
|
} else if (type === 'enter') {
|
|
// The pre-update object was readable on a query-match-gain enter but not
|
|
// on an ACL-grant enter. Only send the pre-update body as `original`
|
|
// when the subscriber could read the original object.
|
|
const originalReadable = isOriginalSubscriptionMatched
|
|
? false
|
|
: await this._matchesACL(message.originalParseObject.getACL(), client, requestId);
|
|
if (!originalReadable) {
|
|
localOriginalParseObject = null;
|
|
}
|
|
}
|
|
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 (classSubscriptions) {
|
|
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 = (node: 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 (node === null || typeof node !== 'object') {
|
|
return;
|
|
}
|
|
if (Array.isArray(node)) {
|
|
for (const item of node) {
|
|
checkDepth(item, depth);
|
|
}
|
|
return;
|
|
}
|
|
// Descend into every value so that logical operators ($or/$and/$nor)
|
|
// nested under field-level operators (e.g. $elemMatch, $not) or plain
|
|
// field names are still counted. Only logical operators increase the
|
|
// depth, which preserves the documented meaning of `queryDepth`.
|
|
for (const key of Object.keys(node)) {
|
|
const isLogical = key === '$or' || key === '$and' || key === '$nor';
|
|
if (isLogical && !Array.isArray(node[key])) {
|
|
throw new Parse.Error(Parse.Error.INVALID_QUERY, `${key} must be an array`);
|
|
}
|
|
checkDepth(node[key], isLogical ? depth + 1 : depth);
|
|
}
|
|
};
|
|
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);
|
|
|
|
// If this client already has a subscription registered under this
|
|
// requestId, replace it by tearing down the previous subscription before
|
|
// creating the new one. The client-side metadata map is keyed only by
|
|
// requestId, so a duplicate `subscribe` frame would otherwise overwrite it
|
|
// while the previous Subscription stays in the server-wide map, leaking it
|
|
// for the lifetime of the process (disconnect cleanup only walks the
|
|
// surviving client metadata and never reaches the orphaned subscription).
|
|
const previousSubscriptionInfo = client.getSubscriptionInfo(request.requestId);
|
|
if (previousSubscriptionInfo) {
|
|
const previousSubscription = previousSubscriptionInfo.subscription;
|
|
previousSubscription.deleteClientSubscription(parseWebsocket.clientId, request.requestId);
|
|
const previousClassSubscriptions = this.subscriptions.get(previousSubscription.className);
|
|
if (previousClassSubscriptions) {
|
|
if (!previousSubscription.hasSubscribingClient()) {
|
|
previousClassSubscriptions.delete(previousSubscription.hash);
|
|
}
|
|
if (previousClassSubscriptions.size === 0) {
|
|
this.subscriptions.delete(previousSubscription.className);
|
|
}
|
|
}
|
|
}
|
|
|
|
// 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 (classSubscriptions) {
|
|
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 };
|