make postgres idempotency function callable

Author: cbaker6

spec/Idempotency.spec.js

@@ -11,8 +11,8 @@ describe('Idempotency', () => {
   /** Enable TTL expiration simulated by removing entry instead of waiting for MongoDB TTL monitor which
    runs only every 60s, so it can take up to 119s until entry removal - ain't nobody got time for that */
   const SIMULATE_TTL = true;
-  const ttl = 30;
-  const maxTimeOut = 130000;
+  const ttl = 2;
+  const maxTimeOut = 4000;
 
   // Helpers
   async function deleteRequestEntry(reqId) {
@@ -93,6 +93,29 @@ describe('Idempotency', () => {
     expect(counter).toBe(2);
   });
 
+  it_only_db('postgres')('should delete request entry when postgress ttl function is called', async () => {
+    const client = Config.get(Parse.applicationId).database.adapter._client;
+    let counter = 0;
+    Parse.Cloud.define('myFunction', () => {
+      counter++;
+    });
+    const params = {
+      method: 'POST',
+      url: 'http://localhost:8378/1/functions/myFunction',
+      headers: {
+        'X-Parse-Application-Id': Parse.applicationId,
+        'X-Parse-Master-Key': Parse.masterKey,
+        'X-Parse-Request-Id': 'abc-123',
+      },
+    };
+    await expectAsync(request(params)).toBeResolved();
+    await expectAsync(request(params)).toBeRejected();
+    await new Promise(resolve => setTimeout(resolve, maxTimeOut));
+    await client.one('SELECT idempodency_delete_old_rows()');
+    await expectAsync(request(params)).toBeResolved();
+    expect(counter).toBe(2);
+  });
+
   it('should enforce idempotency for cloud code jobs', async () => {
     let counter = 0;
     Parse.Cloud.job('myJob', () => {

spec/PostgresStorageAdapter.spec.js

@@ -432,11 +432,13 @@ describe_only_db('postgres')('PostgresStorageAdapter', () => {
 
   it('Idempotency class should have function', async () => {
     await reconfigureServer();
-    const config = Config.get('test');
-    const adapter = config.database.adapter;
+    const adapter = Config.get('test').database.adapter;
     const client = adapter._client;
-    const foundFunction = await client.one("SELECT format('%I.%I(%s)', ns.nspname, p.proname, oidvectortypes(p.proargtypes)) FROM pg_proc p INNER JOIN pg_namespace ns ON (p.pronamespace = ns.oid) WHERE p.proname = 'idempodency_delete_old_rows'");
+    const qs = "SELECT format('%I.%I(%s)', ns.nspname, p.proname, oidvectortypes(p.proargtypes)) FROM pg_proc p INNER JOIN pg_namespace ns ON (p.pronamespace = ns.oid) WHERE p.proname = 'idempodency_delete_old_rows'";
+    const foundFunction = await client.one(qs);
     expect(foundFunction.format).toBe("public.idempodency_delete_old_rows()");
+    await adapter.deleteIdempodencyFunction();
+    await client.none(qs);
   });
 });
 

src/Adapters/Storage/Postgres/PostgresStorageAdapter.js

@@ -886,7 +886,7 @@ export class PostgresStorageAdapter implements StorageAdapter {
   }
 
   async classExists(name: string) {
-    return this._client.one(
+    await this._client.one(
       'SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_name = $1)',
       [name],
       a => a.exists
@@ -964,7 +964,6 @@ export class PostgresStorageAdapter implements StorageAdapter {
       );
     });
     this._notifySchemaChange();
-    return Promise.resolve();
   }
 
   async createClass(className: string, schema: SchemaType, conn: ?any) {
@@ -1030,7 +1029,7 @@ export class PostgresStorageAdapter implements StorageAdapter {
     const qs = `CREATE TABLE IF NOT EXISTS $1:name (${patternsArray.join()})`;
     const values = [className, ...valuesArray];
 
-    return conn.task('create-table', async t => {
+    await conn.task('create-table', async t => {
       try {
         await t.none(qs, values);
       } catch (error) {
@@ -1953,7 +1952,7 @@ export class PostgresStorageAdapter implements StorageAdapter {
     const constraintName = `${className}_unique_${fieldNames.sort().join('_')}`;
     const constraintPatterns = fieldNames.map((fieldName, index) => `$${index + 3}:name`);
     const qs = `CREATE UNIQUE INDEX IF NOT EXISTS $2:name ON $1:name(${constraintPatterns.join()})`;
-    return this._client.none(qs, [className, constraintName, ...fieldNames]).catch(error => {
+    await this._client.none(qs, [className, constraintName, ...fieldNames]).catch(error => {
       if (error.code === PostgresDuplicateRelationError && error.message.includes(constraintName)) {
         // Index already exists. Ignore error.
       } else if (
@@ -2303,7 +2302,7 @@ export class PostgresStorageAdapter implements StorageAdapter {
         .then(() => this.schemaUpgrade(schema.className, schema));
     });
     promises.push(this._listenToSchema());
-    return Promise.all(promises)
+    await Promise.all(promises)
       .then(() => {
         return this._client.tx('perform-initialization', async t => {
           await t.none(sql.misc.jsonObjectSetKeys);
@@ -2421,7 +2420,7 @@ export class PostgresStorageAdapter implements StorageAdapter {
     if (setIdempodencyFunction) {
       await this.ensureIdempodencyFunctionExists(options);
     }
-    return conn.none(qs, [indexNameOptions.name, className, ...fieldNames])
+    await conn.none(qs, [indexNameOptions.name, className, ...fieldNames])
       .catch(error => {
         if (
           error.code === PostgresDuplicateRelationError &&
@@ -2448,7 +2447,7 @@ export class PostgresStorageAdapter implements StorageAdapter {
   ): Promise<any> {
     const conn = options.conn !== undefined ? options.conn : this._client;
     const qs = 'DROP FUNCTION IF EXISTS idempodency_delete_old_rows()';
-    return conn
+    await conn
       .none(qs)
       .catch(error => {
         throw error;
@@ -2460,8 +2459,8 @@ export class PostgresStorageAdapter implements StorageAdapter {
   ): Promise<any> {
     const conn = options.conn !== undefined ? options.conn : this._client;
     const ttlOptions = options.ttl !== undefined ? `${options.ttl} seconds` : '60 seconds';
-    const qs = 'CREATE OR REPLACE FUNCTION idempodency_delete_old_rows() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN DELETE FROM "_Idempotency" WHERE expire < NOW() - INTERVAL $1; RETURN NEW; END; $$;';
-    return conn
+    const qs = 'CREATE OR REPLACE FUNCTION idempodency_delete_old_rows() RETURNS void LANGUAGE plpgsql AS $$ BEGIN DELETE FROM "_Idempotency" WHERE expire < NOW() - INTERVAL $1; END; $$;';
+    await conn
       .none(qs, [ttlOptions])
       .catch(error => {
         if (