feat: Add push provider throttle and push priority processing

README.md

@@ -25,6 +25,9 @@ The official Push Notification adapter for Parse Server. See [Parse Server Push
     - [HTTP/1.1 Legacy Option](#http11-legacy-option)
     - [Firebase Client Error](#firebase-client-error)
   - [Expo Push Options](#expo-push-options)
+- [Push Queue](#push-queue)
+  - [Throttling](#throttling)
+  - [Push Options](#push-options)
 - [Bundled with Parse Server](#bundled-with-parse-server)
 - [Logging](#logging)
 
@@ -181,6 +184,83 @@ expo: {
 
 For more information see the [Expo docs](https://docs.expo.dev/push-notifications/overview/).
 
+## Push Queue
+
+### Throttling
+
+By default, pushes are sent as fast as possible. However, push providers usually throttle their APIs, so that sending too many pushes notifications within a short time may cause the API to reject requests. To address this, push sending can be throttled per provider by adding the `queue` option to the respective push configuration.
+
+| Parameter                | Default    | Optional | Description                                                                                |
+|--------------------------|------------|----------|--------------------------------------------------------------------------------------------|
+| `queue.concurrency`      | `Infinity` | Yes      | The maximum number of pushes to process concurrently.                                      |
+| `queue.intervalCapacity` | `Infinity` | Yes      | The interval capacity, meaning the maximum number of tasks to process in a given interval. |
+| `queue.interval`         | `0`        | Yes      | The interval in milliseconds for the interval capacity.                                    |
+
+Example configuration to throttle the queue to max. 1 push every 100ms, equivalent to max. 10 pushes per second:
+
+```js
+const parseServerOptions = {
+  push: {
+    adapter: new ParsePushAdapter({
+      ios: {
+        // ...
+        queue: {
+          concurrency: 1,
+          intervalCapacity: 1,
+          interval: 100,
+        },
+      }
+    })
+  }
+};
+```
+
+Keep in mind that `concurrency: 1` means that pushes are sent in serial. For example, if sending a request to the push provider takes up to 500ms to complete, then the configuration above may be limited to only 2 pushes per second if every request takes 500ms. To address this, you can send pushes in parallel by setting the concurrency to a value greater than `1`, and increasing `intervalCapacity` and `interval` to fully utilize parallelism.
+
+Example configuration sending pushes in parallel:
+
+```js
+const parseServerOptions = {
+  push: {
+    adapter: new ParsePushAdapter({
+      ios: {
+        // ...
+        queue: {
+          concurrency: 5,
+          intervalCapacity: 5,
+          interval: 500,
+        },
+      }
+    })
+  }
+};
+```
+
+In the example above, pushes will be sent in bursts of 5 at once, with max. 10 pushes within 1s. On a timeline that means at `t=0ms`, 5 pushes will be sent in parallel. If sending the pushes take less than 500ms, then `intervalCapacity` will still limit to 5 pushes within the first 500ms. At `t=500ms` the second interval begins and another max. 5 pushes are sent in parallel. That effectively means a throughput of up to 10 pushes per second.
+
+### Push Options
+
+Each push request may specify the following options for handling in the queue.
+
+| Parameter        | Default    | Optional | Description                                                                                                                                                                                  |
+|------------------|------------|----------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| `queue.ttl`      | `Infinity` | Yes      | The time-to-live of the push in the queue in seconds. If a queued push expires before it is sent to the push provider, it is discarded. Default is `Infinity`, meaning pushes never expire.  |
+| `queue.priority` | `0`        | Yes      | The priority of the push in the queue. When processing the queue, pushes are sent in order of their priority. For example, a push with priority `1` is sent before a push with priority `0`. |
+
+Example push payload:
+
+```js
+pushData = {
+  queue: {
+    // Discard after 10 seconds from queue if push has not been sent to push provider yet
+    ttl: 10,
+    // Send with higher priority than default pushes
+    priority: 1,
+  },
+  data: { alert: 'Hello' }
+};
+```
+
 ## Bundled with Parse Server
 
 Parse Server already comes bundled with a specific version of the push adapter. This installation is only necessary when customizing the push adapter version that should be used by Parse Server. When using a customized version of the push adapter, ensure that it's compatible with the version of Parse Server you are using.

package.json

@@ -29,6 +29,7 @@
     "expo-server-sdk": "3.15.0",
     "firebase-admin": "13.4.0",
     "npmlog": "7.0.1",
+    "p-queue": "8.1.0",
     "parse": "6.1.1",
     "web-push": "3.6.7"
   },

spec/.eslintrc.json

@@ -1,6 +1,6 @@
 {
     "env": {
-        "jasmine": true
+      "jasmine": true
     },
     "globals": {
       "Parse": true

spec/ParsePushAdapter.spec.js

@@ -1,15 +1,16 @@
-import { join } from 'path';
-import log from 'npmlog';
 import apn from '@parse/node-apn';
-import ParsePushAdapterPackage, { ParsePushAdapter as _ParsePushAdapter, APNS as _APNS, GCM as _GCM, WEB as _WEB, EXPO as _EXPO, utils } from '../src/index.js';
-const ParsePushAdapter = _ParsePushAdapter;
-import { randomString } from '../src/PushAdapterUtils.js';
-import MockAPNProvider from './MockAPNProvider.js';
+import log from 'npmlog';
+import { join } from 'path';
 import APNS from '../src/APNS.js';
+import EXPO from '../src/EXPO.js';
+import FCM from '../src/FCM.js';
 import GCM from '../src/GCM.js';
+import ParsePushAdapterPackage, { APNS as _APNS, EXPO as _EXPO, GCM as _GCM, ParsePushAdapter as _ParsePushAdapter, WEB as _WEB, utils } from '../src/index.js';
+import { randomString } from '../src/PushAdapterUtils.js';
 import WEB from '../src/WEB.js';
-import FCM from '../src/FCM.js';
-import EXPO from '../src/EXPO.js';
+import { wait } from './helper.js';
+import MockAPNProvider from './MockAPNProvider.js';
+const ParsePushAdapter = _ParsePushAdapter;
 
 describe('ParsePushAdapter', () => {
 
@@ -642,6 +643,99 @@ describe('ParsePushAdapter', () => {
     });
   });
 
+
+  it('throttles push sends per provider', async () => {
+    const pushConfig = {
+      android: {
+        senderId: 'id',
+        apiKey: 'key',
+        queue: {
+          concurrency: 1,
+          intervalCapacity: 1,
+          interval: 1_000,
+        },
+      },
+    };
+    const parsePushAdapter = new ParsePushAdapter(pushConfig);
+    const times = [];
+    parsePushAdapter.senderMap['android'].send = jasmine.createSpy('send').and.callFake(() => {
+      times.push(Date.now());
+      return Promise.resolve([]);
+    });
+    const installs = [{ deviceType: 'android', deviceToken: 'token' }];
+    await Promise.all([
+      parsePushAdapter.send({}, installs),
+      parsePushAdapter.send({}, installs),
+    ]);
+    expect(times.length).toBe(2);
+    expect(times[1] - times[0]).toBeGreaterThanOrEqual(900);
+    expect(times[1] - times[0]).toBeLessThanOrEqual(1100);
+  });
+
+  it('skips queued pushes after ttl expires', async () => {
+    const pushConfig = {
+      android: {
+        senderId: 'id',
+        apiKey: 'key',
+        queue: {
+          concurrency: 1,
+          intervalCapacity: 1,
+          interval: 1_000,
+        },
+      },
+    };
+    const parsePushAdapter = new ParsePushAdapter(pushConfig);
+    parsePushAdapter.senderMap['android'].send = jasmine.createSpy('send').and.callFake(async () => {
+      await wait(1_200);
+      return [];
+    });
+    const installs = [{ deviceType: 'android', deviceToken: 'token' }];
+    await Promise.all([
+      parsePushAdapter.send({}, installs),
+      parsePushAdapter.send({ queue: { ttl: 1 } }, installs)
+    ]);
+    expect(parsePushAdapter.senderMap['android'].send.calls.count()).toBe(1);
+  });
+
+  it('sends higher priority pushes before lower priority ones', async () => {
+    const pushConfig = {
+      android: {
+        senderId: 'id',
+        apiKey: 'key',
+        queue: {
+          concurrency: 1,
+          intervalCapacity: 1,
+          interval: 1_000,
+        },
+      },
+    };
+    const parsePushAdapter = new ParsePushAdapter(pushConfig);
+    const callOrder = [];
+    parsePushAdapter.senderMap['android'].send = jasmine.createSpy('send').and.callFake(async (data) => {
+      callOrder.push(data.id);
+      await wait(100);
+      return [];
+    });
+    const installs = [{ deviceType: 'android', deviceToken: 'token' }];
+
+    // Block queue with task so that the queue scheduler doesn't start processing enqueued items
+    // immediately; afterwards the scheduler picks the next enqueued item according to priority;
+    const pBlock = parsePushAdapter.queues.android.enqueue({ task: () => wait(500) });
+    // Wait to ensure block item in queue has started
+    await wait(100);
+
+    await Promise.all([
+      pBlock,
+      parsePushAdapter.send({ id: 3, queue: { priority: 3 }}, installs),
+      parsePushAdapter.send({ id: 4, queue: { priority: 4 }}, installs),
+      parsePushAdapter.send({ id: 2, queue: { priority: 2 }}, installs),
+      parsePushAdapter.send({ id: 0, queue: { priority: 0 }}, installs),
+      parsePushAdapter.send({ id: 1, queue: { priority: 1 }}, installs),
+    ]);
+    expect(callOrder).toEqual([4, 3, 2, 1, 0]);
+  });
+
+
   it('random string throws with size <=0', () => {
     expect(() => randomString(0)).toThrow();
   });