Merge branch 'master' into schema-reduction

Author: dplewis

package-lock.json

@@ -47,17 +47,17 @@
     "mime": "2.5.2",
     "mongodb": "3.6.6",
     "mustache": "4.2.0",
-    "parse": "3.1.0",
+    "parse": "3.2.0",
     "pg-monitor": "1.4.1",
-    "pg-promise": "10.9.2",
+    "pg-promise": "10.10.1",
     "pluralize": "8.0.0",
-    "redis": "3.0.2",
+    "redis": "3.1.1",
     "semver": "7.3.4",
     "subscriptions-transport-ws": "0.9.18",
     "tv4": "1.3.0",
     "uuid": "8.3.2",
     "winston": "3.3.3",
-    "winston-daily-rotate-file": "4.5.1",
+    "winston-daily-rotate-file": "4.5.2",
     "ws": "7.4.4"
   },
   "devDependencies": {

package.json

@@ -645,6 +645,69 @@ describe('ParseLiveQuery', function () {
     await object.save();
   });
 
+  it('LiveQuery with ACL', async () => {
+    await reconfigureServer({
+      liveQuery: {
+        classNames: ['Chat'],
+      },
+      startLiveQueryServer: true,
+      verbose: false,
+      silent: true,
+    });
+    const user = new Parse.User();
+    user.setUsername('username');
+    user.setPassword('password');
+    await user.signUp();
+
+    const calls = {
+      beforeConnect(req) {
+        expect(req.event).toBe('connect');
+        expect(req.clients).toBe(0);
+        expect(req.subscriptions).toBe(0);
+        expect(req.useMasterKey).toBe(false);
+        expect(req.installationId).toBeDefined();
+        expect(req.client).toBeDefined();
+      },
+      beforeSubscribe(req) {
+        expect(req.op).toBe('subscribe');
+        expect(req.requestId).toBe(1);
+        expect(req.query).toBeDefined();
+        expect(req.user).toBeDefined();
+      },
+      afterLiveQueryEvent(req) {
+        expect(req.user).toBeDefined();
+        expect(req.object.get('foo')).toBe('bar');
+      },
+      create(object) {
+        expect(object.get('foo')).toBe('bar');
+      },
+      delete(object) {
+        expect(object.get('foo')).toBe('bar');
+      },
+    };
+    for (const key in calls) {
+      spyOn(calls, key).and.callThrough();
+    }
+    Parse.Cloud.beforeConnect(calls.beforeConnect);
+    Parse.Cloud.beforeSubscribe('Chat', calls.beforeSubscribe);
+    Parse.Cloud.afterLiveQueryEvent('Chat', calls.afterLiveQueryEvent);
+
+    const chatQuery = new Parse.Query('Chat');
+    const subscription = await chatQuery.subscribe();
+    subscription.on('create', calls.create);
+    subscription.on('delete', calls.delete);
+    const object = new Parse.Object('Chat');
+    const acl = new Parse.ACL(user);
+    object.setACL(acl);
+    object.set({ foo: 'bar' });
+    await object.save();
+    await object.destroy();
+    await new Promise(resolve => setTimeout(resolve, 200));
+    for (const key in calls) {
+      expect(calls[key]).toHaveBeenCalled();
+    }
+  });
+
   it('handle invalid websocket payload length', async done => {
     await reconfigureServer({
       liveQuery: {

spec/ParseLiveQuery.spec.js

@@ -1056,29 +1056,45 @@ export class PostgresStorageAdapter implements StorageAdapter {
     conn = conn || this._client;
     const self = this;
 
-    await conn.tx('schema-upgrade', async t => {
+    await conn.task('schema-upgrade', async t => {
       const columns = await t.map(
         'SELECT column_name FROM information_schema.columns WHERE table_name = $<className>',
         { className },
         a => a.column_name
       );
       const newColumns = Object.keys(schema.fields)
         .filter(item => columns.indexOf(item) === -1)
-        .map(fieldName =>
-          self.addFieldIfNotExists(className, fieldName, schema.fields[fieldName], t)
-        );
+        .map(fieldName => self.addFieldIfNotExists(className, fieldName, schema.fields[fieldName]));
 
       await t.batch(newColumns);
     });
   }
 
-  async addFieldIfNotExists(className: string, fieldName: string, type: any, conn: any) {
+  async addFieldIfNotExists(className: string, fieldName: string, type: any) {
     // TODO: Must be revised for invalid logic...
     debug('addFieldIfNotExists');
-    const t = (conn && conn.t ? conn.t : conn) || this._client;
     const self = this;
-    if (type.type !== 'Relation') {
-      try {
+    await this._client.tx('add-field-if-not-exists', async t => {
+      if (type.type !== 'Relation') {
+        try {
+          await t.none(
+            'ALTER TABLE $<className:name> ADD COLUMN IF NOT EXISTS $<fieldName:name> $<postgresType:raw>',
+            {
+              className,
+              fieldName,
+              postgresType: parseTypeToPostgresType(type),
+            }
+          );
+        } catch (error) {
+          if (error.code === PostgresRelationDoesNotExistError) {
+            return self.createClass(className, { fields: { [fieldName]: type } }, t);
+          }
+          if (error.code !== PostgresDuplicateColumnError) {
+            throw error;
+          }
+          // Column already exists, created by other request. Carry on to see if it's the right type.
+        }
+      } else {
         await t.none(
           'ALTER TABLE $<className:name> ADD COLUMN IF NOT EXISTS $<fieldName:name> $<postgresType:raw>',
           {

src/Adapters/Storage/Postgres/PostgresStorageAdapter.js

@@ -170,8 +170,10 @@ class ParseLiveQueryServer {
             };
             const trigger = getTrigger(className, 'afterEvent', Parse.applicationId);
             if (trigger) {
-              const auth = await this.getAuthForSessionToken(res.sessionToken);
-              res.user = auth.user;
+              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);
               }
@@ -317,8 +319,10 @@ class ParseLiveQueryServer {
               if (res.original) {
                 res.original = Parse.Object.fromJSON(res.original);
               }
-              const auth = await this.getAuthForSessionToken(res.sessionToken);
-              res.user = auth.user;
+              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) {
@@ -579,6 +583,24 @@ class ParseLiveQueryServer {
       });
   }
 
+  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;
+  }
+
   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) {
@@ -631,8 +653,10 @@ class ParseLiveQueryServer {
       };
       const trigger = getTrigger('@Connect', 'beforeConnect', Parse.applicationId);
       if (trigger) {
-        const auth = await this.getAuthForSessionToken(req.sessionToken);
-        req.user = auth.user;
+        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;
@@ -690,8 +714,10 @@ class ParseLiveQueryServer {
     try {
       const trigger = getTrigger(className, 'beforeSubscribe', Parse.applicationId);
       if (trigger) {
-        const auth = await this.getAuthForSessionToken(request.sessionToken);
-        request.user = auth.user;
+        const auth = await this.getAuthFromClient(client, request.requestId, request.sessionToken);
+        if (auth && auth.user) {
+          request.user = auth.user;
+        }
 
         const parseQuery = new Parse.Query(className);
         parseQuery.withJSON(request.query);