remove improvements

Author: dplewis

spec/PostgresStorageAdapter.spec.js

@@ -385,14 +385,23 @@ describe_only_db('postgres')('PostgresStorageAdapter', () => {
     const { database } = Config.get(Parse.applicationId);
     const { adapter } = database;
 
-    spyOn(adapter, 'watch');
     spyOn(adapter, '_onchange');
-    const schema = await database.loadSchema();
-    // Create a valid class
-    await schema.validateObject('Stuff', { foo: 'bar' });
-    await new Promise(resolve => setTimeout(resolve, 500));
 
-    expect(adapter.watch).toHaveBeenCalled();
+    const otherInstance = new PostgresStorageAdapter({ uri: databaseURI });
+    otherInstance._listenToSchema();
+
+    await otherInstance.createClass('Stuff', {
+      className: 'Stuff',
+      fields: {
+        objectId: { type: 'String' },
+        createdAt: { type: 'Date' },
+        updatedAt: { type: 'Date' },
+        _rperm: { type: 'Array' },
+        _wperm: { type: 'Array' },
+      },
+      classLevelPermissions: undefined,
+    });
+    await new Promise(resolve => setTimeout(resolve, 500));
     expect(adapter._onchange).toHaveBeenCalled();
   });
 });

src/Adapters/Storage/Postgres/PostgresStorageAdapter.js

@@ -4,6 +4,7 @@ import { createClient } from './PostgresClient';
 import Parse from 'parse/node';
 // @flow-disable-next
 import _ from 'lodash';
+import { v4 as uuidv4 } from 'uuid';
 import sql from './sql';
 
 const PostgresRelationDoesNotExistError = '42P01';
@@ -801,6 +802,7 @@ export class PostgresStorageAdapter implements StorageAdapter {
   _onchange: any;
   _pgp: any;
   _stream: any;
+  _uuid: any;
 
   constructor({ uri, collectionPrefix = '', databaseOptions }: any) {
     this._collectionPrefix = collectionPrefix;
@@ -809,6 +811,7 @@ export class PostgresStorageAdapter implements StorageAdapter {
     this._onchange = () => {};
     this._pgp = pgp;
     this.canSortOnJoinTables = false;
+    this.uuid = uuidv4();
   }
 
   watch(callback: () => void): void {
@@ -835,11 +838,26 @@ export class PostgresStorageAdapter implements StorageAdapter {
     this._client.$pool.end();
   }
 
+  async _listenToSchema() {
+    if (!this._stream) {
+      this._stream = await this._client.connect({ direct: true });
+      this._stream.client.on('notification', data => {
+        const payload = JSON.parse(data.payload);
+        if (payload.senderId !== this.uuid) {
+          this._onchange();
+        }
+      });
+      await this._stream.none('LISTEN $1~', 'schema.change');
+    }
+  }
+
   _notifySchemaChange() {
     if (this._stream) {
-      this._stream.none('NOTIFY $1~, $2', ['schema.change', '']).catch(error => {
-        console.log('Failed to Notify:', error); // unlikely to ever happen
-      });
+      this._stream
+        .none('NOTIFY $1~, $2', ['schema.change', { senderId: this.uuid }])
+        .catch(error => {
+          console.log('Failed to Notify:', error); // unlikely to ever happen
+        });
     }
   }
 
@@ -948,7 +966,7 @@ export class PostgresStorageAdapter implements StorageAdapter {
 
   async createClass(className: string, schema: SchemaType, conn: ?any) {
     conn = conn || this._client;
-    return conn
+    const parseSchema = await conn
       .tx('create-class', async t => {
         await this.createTable(className, schema, t);
         await t.none(
@@ -964,6 +982,8 @@ export class PostgresStorageAdapter implements StorageAdapter {
         }
         throw err;
       });
+    this._notifySchemaChange();
+    return parseSchema;
   }
 
   // Just create a table, do not insert in schema
@@ -2270,11 +2290,6 @@ export class PostgresStorageAdapter implements StorageAdapter {
   async performInitialization({ VolatileClassesSchemas }: any) {
     // TODO: This method needs to be rewritten to make proper use of connections (@vitaly-t)
     debug('performInitialization');
-    if (!this._stream) {
-      this._stream = await this._client.connect({ direct: true });
-      this._stream.client.on('notification', () => this._onchange());
-      await this._stream.none('LISTEN $1~', 'schema.change');
-    }
     const promises = VolatileClassesSchemas.map(schema => {
       return this.createTable(schema.className, schema)
         .catch(err => {
@@ -2288,6 +2303,7 @@ export class PostgresStorageAdapter implements StorageAdapter {
         })
         .then(() => this.schemaUpgrade(schema.className, schema));
     });
+    promises.push(this._listenToSchema());
     return Promise.all(promises)
       .then(() => {
         return this._client.tx('perform-initialization', async t => {

src/Controllers/SchemaController.js

@@ -808,19 +808,7 @@ export default class SchemaController {
           className,
         })
       )
-      .then(adapterSchema => {
-        const parseSchema = convertAdapterSchemaToParseSchema(adapterSchema);
-        this._cache.allClasses = this._cache.allClasses || [];
-        const index = this._cache.allClasses.findIndex(
-          cached => cached.className === parseSchema.className
-        );
-        if (index >= 0) {
-          this._cache.allClasses[index] = parseSchema;
-        } else {
-          this._cache.allClasses.push(parseSchema);
-        }
-        return parseSchema;
-      })
+      .then(convertAdapterSchemaToParseSchema)
       .catch(error => {
         if (error && error.code === Parse.Error.DUPLICATE_VALUE) {
           throw new Parse.Error(
@@ -946,7 +934,7 @@ export default class SchemaController {
     return (
       this.addClassIfNotExists(className)
         // The schema update succeeded. Reload the schema
-        .then(() => this.reloadData())
+        .then(() => this.reloadData({ clearCache: true }))
         .catch(() => {
           // The schema update failed. This can be okay - it might
           // have failed because there's a race condition and a different
@@ -1132,12 +1120,6 @@ export default class SchemaController {
         return Promise.resolve();
       })
       .then(() => {
-        const cached = (this._cache.allClasses || []).find(
-          schema => schema.className === className
-        );
-        if (cached && !cached.fields[fieldName]) {
-          cached.fields[fieldName] = type;
-        }
         return {
           className,
           fieldName,
@@ -1230,7 +1212,7 @@ export default class SchemaController {
   async validateObject(className: string, object: any, query: any) {
     let geocount = 0;
     const schema = await this.enforceClassExists(className);
-    const results = [];
+    const promises = [];
 
     for (const fieldName in object) {
       if (object[fieldName] === undefined) {
@@ -1257,12 +1239,13 @@ export default class SchemaController {
         // Every object has ACL implicitly.
         continue;
       }
-      results.push(await schema.enforceFieldExists(className, fieldName, expected));
+      promises.push(schema.enforceFieldExists(className, fieldName, expected));
     }
+    const results = await Promise.all(promises);
     const enforceFields = results.filter(result => !!result);
 
     if (enforceFields.length !== 0) {
-      await this.reloadData();
+      await this.reloadData({ clearCache: true });
     }
     this.ensureFields(enforceFields);