Add push notification callbacks to interaction processing.

Author: dlongley

lib/interactions.js

@@ -3,7 +3,7 @@
  */
 import * as bedrock from '@bedrock/core';
 import * as schemas from '../schemas/bedrock-profile-http.js';
-import {poll, pollers} from '@bedrock/notify';
+import {poll, pollers, push} from '@bedrock/notify';
 import {agent} from '@bedrock/https-agent';
 import {asyncHandler} from '@bedrock/express';
 import {ensureAuthenticated} from '@bedrock/passport';
@@ -16,6 +16,12 @@ const {config, util: {BedrockError}} = bedrock;
 let DEFINITIONS_BY_TYPE_MAP;
 let DEFINITIONS_BY_ID_MAP;
 
+// use a TTL of 1 second to account for the case where a push notification
+// isn't received by the same instance that the client hits, but prevent
+// requests from triggering a hit to the workflow service backend more
+// frequently than 1 second
+const POLL_TTL = 1000;
+
 bedrock.events.on('bedrock.init', () => {
   const cfg = config['profile-http'];
 
@@ -65,9 +71,13 @@ bedrock.events.on('bedrock-express.configure.routes', app => {
   const interactionsPath = '/interactions';
   const routes = {
     interactions: interactionsPath,
-    interaction: `${interactionsPath}/:localInteractionId/:localExchangeId`
+    interaction: `${interactionsPath}/:localInteractionId/:localExchangeId`,
+    callback: `${interactionsPath}/:localInteractionId/callbacks/:pushToken`
   };
 
+  // base URL for server
+  const {baseUri} = bedrock.config.server;
+
   // create an interaction to exchange VCs
   app.post(
     routes.interactions,
@@ -88,6 +98,15 @@ bedrock.events.on('bedrock-express.configure.routes', app => {
         });
       }
 
+      // create a push token
+      const {token} = await push.createPushToken({event: 'exchangeUpdated'});
+
+      // compute callback URL
+      const {localInteractionId} = definition;
+      const callbackUrl =
+        `${baseUri}${interactionsPath}/${localInteractionId}` +
+        `/callbacks/${token}`;
+
       // create exchange with given variables
       const exchange = {
         // FIXME: use `expires` instead of now-deprecated `ttl`
@@ -96,13 +115,15 @@ bedrock.events.on('bedrock-express.configure.routes', app => {
         // template variables
         variables: {
           ...variables,
+          callback: {
+            url: callbackUrl
+          },
           accountId
         }
       };
       const capability = definition.zcaps.get('readWriteExchanges');
       const response = await zcapClient.write({json: exchange, capability});
       const exchangeId = response.headers.get('location');
-      const {localInteractionId} = definition;
       // reuse `localExchangeId` in path
       const localExchangeId = exchangeId.slice(exchangeId.lastIndexOf('/'));
       const id = `${config.server.baseUri}/${routes.interactions}/` +
@@ -123,14 +144,7 @@ bedrock.events.on('bedrock-express.configure.routes', app => {
       } = req;
 
       // get interaction definition
-      const definition = DEFINITIONS_BY_ID_MAP.get(localInteractionId);
-      if(!definition) {
-        throw new BedrockError(
-          `Interaction type for "${localInteractionId}" not found.`, {
-            name: 'NotFoundError',
-            details: {httpStatusCode: 404, public: true}
-          });
-      }
+      const definition = _getInteractionDefinition({localInteractionId});
 
       // determine full exchange ID based on related capability
       const capability = definition.zcaps.get('readWriteExchanges');
@@ -162,36 +176,79 @@ bedrock.events.on('bedrock-express.configure.routes', app => {
       // poll the exchange...
       const result = await poll({
         id: exchangeId,
-        poller: pollers.createExchangePoller({
-          zcapClient,
-          capability,
-          filterExchange({exchange/*, previousPollResult*/}) {
-            // ensure `accountId` matches exchange variables
-            if(exchange?.variables.accountId !== accountId) {
-              throw new BedrockError('Not authorized.', {
-                name: 'NotAllowedError',
-                details: {httpStatusCode: 403, public: true}
-              });
-            }
-            // return only information that should be accessible to client
-            return {
-              exchange
-              // FIXME: filter info once final step name and info is determined
-              /*
-              exchange: {
-                state: exchange.state,
-                result: exchange.variables.results?.finish
-              }*/
-            };
-          }
-        }),
-        // set a TTL of 1 seconds to account for the case where a push
-        // notification isn't received by the same instance that the client
-        // hits, but prevent requests from triggering a hit to the backend more
-        // frequently than 1 second
-        ttl: 1000
+        poller: _createExchangePoller({accountId, capability}),
+        ttl: POLL_TTL
       });
 
       res.json(result);
     }));
+
+  // push event handler
+  app.post(
+    routes.callback,
+    push.createVerifyPushTokenMiddleware({event: 'exchangeUpdated'}),
+    asyncHandler(async (req, res) => {
+      const {event: {data: {exchangeId: id}}} = req.body;
+      const {localInteractionId} = req.params;
+
+      // get interaction definition
+      const definition = _getInteractionDefinition({localInteractionId});
+
+      // get capability for fetching exchange and verify its invocation target
+      // matches the exchange ID passed
+      const capability = definition.zcaps.get('readWriteExchanges');
+      if(!id.startsWith(capability.invocationTarget)) {
+        throw new BedrockError('Not authorized.', {
+          name: 'NotAllowedError',
+          details: {httpStatusCode: 403, public: true}
+        });
+      }
+
+      // poll (and clear cache)
+      await poll({
+        id,
+        poller: _createExchangePoller({capability}),
+        ttl: POLL_TTL,
+        useCache: false
+      });
+      res.sendStatus(204);
+    }));
 });
+
+function _createExchangePoller({accountId, capability}) {
+  return pollers.createExchangePoller({
+    zcapClient,
+    capability,
+    filterExchange({exchange/*, previousPollResult*/}) {
+      // if `accountId` given, ensure it matches exchange variables
+      if(accountId && exchange?.variables.accountId !== accountId) {
+        throw new BedrockError('Not authorized.', {
+          name: 'NotAllowedError',
+          details: {httpStatusCode: 403, public: true}
+        });
+      }
+      // return only information that should be accessible to client
+      return {
+        exchange
+        // FIXME: filter info once final step name and info is determined
+        /*
+        exchange: {
+          state: exchange.state,
+          result: exchange.variables.results?.finish
+        }*/
+      };
+    }
+  });
+}
+
+function _getInteractionDefinition({localInteractionId}) {
+  const definition = DEFINITIONS_BY_ID_MAP.get(localInteractionId);
+  if(!definition) {
+    throw new BedrockError(
+      `Interaction type for "${localInteractionId}" not found.`, {
+        name: 'NotFoundError',
+        details: {httpStatusCode: 404, public: true}
+      });
+  }
+  return definition;
+}

test/test.config.js

@@ -46,3 +46,8 @@ config['profile-http'].interactions.types.test = {
     readWriteExchanges: '{}'
   }
 };
+// test hmac key for push token feature; required for `interactions`
+config.notify.push.hmacKey = {
+  id: 'urn:test:hmacKey',
+  secretKeyMultibase: 'uogHy02QDNPX4GID7dGUSGuYQ_Gv0WOIcpmTuKgt1ZNz7_4'
+};