Improve webhooks experience (#4396)

* Improve webhooks experience

* Fix changesets

.changeset/friendly-singers-call.md

@@ -0,0 +1,9 @@
+---
+"@graphql-mesh/cli": minor
+"@omnigraph/json-schema": minor
+"@omnigraph/openapi": minor
+"@graphql-mesh/types": minor
+"@graphql-mesh/utils": minor
+---
+
+Drop webhook plugin and automatically handle webhooks. See the documentation for more information

examples/json-schema-subscriptions/.meshrc.yml

@@ -19,7 +19,7 @@ sources:
             responseSample: ./todo.json
           - type: Subscription
             field: todoAdded
-            pubsubTopic: todoAdded
+            pubsubTopic: webhook:post:/webhooks/todo_added
             responseSample: ./todo.json
 
 documents:
@@ -29,10 +29,6 @@ additionalTypeDefs: |
   directive @live on QUERY
 
 plugins:
-  - webhook:
-      path: /webhooks/todo_added
-      method: POST
-      pubsubTopic: todoAdded
   - liveQuery:
       invalidations:
         - field: Mutation.addTodo

examples/json-schema-subscriptions/package.json

@@ -11,7 +11,6 @@
     "@graphql-mesh/cli": "0.76.2",
     "@graphql-mesh/json-schema": "0.34.1",
     "@graphql-mesh/plugin-live-query": "0.1.14",
-    "@graphql-mesh/plugin-webhook": "0.0.11",
     "graphql": "16.6.0",
     "body-parser": "1.20.0",
     "express": "4.18.1",

examples/openapi-subscriptions/.meshrc.yml

@@ -8,10 +8,5 @@ sources:
 documents:
   - ./example-queries/**/*.graphql
 
-plugins:
-  - webhook:
-      path: /callback
-      pubsubTopic: http://localhost:4000/callback
-
 serve:
   port: 4000

examples/openapi-subscriptions/api.js

@@ -1,19 +1,22 @@
 const express = require('express');
 const { fetch } = require('@whatwg-node/fetch');
 const bodyParser = require('body-parser');
+const urljoin = require('url-join');
 
 const app = express();
 app.use(bodyParser.json());
 
 app.post('/streams', async (req, res) => {
   const { callbackUrl } = req.body;
+  const subscriptionId = Date.now().toString();
   setInterval(() => {
     const body = JSON.stringify({
       timestamp: new Date().toJSON(),
       userData: 'RANDOM_DATA',
     });
-    console.info('Webhook ping -> ', callbackUrl, body);
-    fetch(callbackUrl, {
+    const fullCallbackUrl = urljoin(callbackUrl, subscriptionId);
+    console.info('Webhook ping -> ', fullCallbackUrl, body);
+    fetch(fullCallbackUrl, {
       method: 'POST',
       headers: {
         'Content-Type': 'application/json',
@@ -22,7 +25,7 @@ app.post('/streams', async (req, res) => {
     }).catch(console.log);
   }, 1000);
   res.json({
-    subscriptionId: Date.now().toString(),
+    subscriptionId,
   });
 });
 

examples/openapi-subscriptions/example-queries/listenWebhook.mutation.graphql

@@ -1,5 +1,5 @@
-subscription {
-  postRequestBodyCallbackUrlData(streamsInput: { callbackUrl: "http://localhost:4000/callback" }) {
+subscription listenWebhook($subscriptionId: String!) {
+  onData(subscriptionId: $subscriptionId) {
     userData
   }
 }

examples/openapi-subscriptions/example-queries/startWebhook.subscription.graphql

@@ -1,5 +1,5 @@
-mutation {
-  postStreams(streamsInput: { callbackUrl: "http://localhost:4000/callback" }) {
+mutation startWebhook {
+  post_streams(input: { callbackUrl: "http://localhost:4000/callback" }) {
     subscriptionId
   }
 }

examples/openapi-subscriptions/openapi.yml

@@ -25,6 +25,7 @@ paths:
           content:
             application/json:
               schema:
+                title: SubscriptionInfo
                 description: subscription information
                 required:
                   - subscriptionId
@@ -38,13 +39,14 @@ paths:
         onData:
           # when data is sent, it will be sent to the `callbackUrl` provided
           # when making the subscription PLUS the suffix `/data`
-          '{$request.body#/callbackUrl}':
+          '/callback/{$response.body#/subscriptionId}':
             post:
               requestBody:
                 description: subscription payload
                 content:
                   application/json:
                     schema:
+                      title: SubscriptionPayload
                       type: object
                       properties:
                         timestamp:

packages/cli/src/commands/serve/serve.ts

@@ -219,6 +219,29 @@ export async function serveMesh(
     app.get('/healthcheck', (_req, res) => res.sendStatus(200));
     app.get('/readiness', (_req, res) => res.sendStatus(readyFlag ? 200 : 500));
 
+    app.use((req, res, next) => {
+      mesh$
+        .then(async ({ pubsub }) => {
+          for (const eventName of pubsub.getEventNames()) {
+            if (eventName === `webhook:${req.method.toLowerCase()}:${req.path}`) {
+              const chunks: Buffer[] = [];
+              for await (const chunk of req) {
+                chunks.push(chunk);
+              }
+              const body = Buffer.concat(chunks).toString('utf-8');
+              logger.debug(`Received webhook request for ${req.path}`, body);
+              pubsub.publish(eventName, req.headers['content-type'] === 'application/json' ? JSON.parse(body) : body);
+              res.status(200).send('OK');
+              return;
+            }
+          }
+          next();
+        })
+        .catch(() => {
+          next();
+        });
+    });
+
     if (staticFiles) {
       app.use(express.static(staticFiles));
       const indexPath = path.join(baseDir, staticFiles, 'index.html');

packages/loaders/json-schema/src/addExecutionLogicToComposer.ts

@@ -123,7 +123,12 @@ export async function addExecutionLogicToComposer(
           return new GraphQLError(`You should have PubSub defined in either the config or the context!`);
         }
         const interpolationData = { root, args, context, info, env: process.env };
-        const pubsubTopic = stringInterpolator.parse(operationConfig.pubsubTopic, interpolationData);
+        let pubsubTopic: string = stringInterpolator.parse(operationConfig.pubsubTopic, interpolationData);
+        if (pubsubTopic.startsWith('webhook:')) {
+          const [, expectedMethod, expectedUrl] = pubsubTopic.split(':');
+          const expectedPath = new URL(expectedUrl, 'http://localhost').pathname;
+          pubsubTopic = `webhook:${expectedMethod}:${expectedPath}`;
+        }
         operationLogger.debug(`=> Subscribing to pubSubTopic: ${pubsubTopic}`);
         return pubsub.asyncIterator(pubsubTopic);
       };

packages/loaders/openapi/src/getJSONSchemaOptionsFromOpenAPIOptions.ts

@@ -462,19 +462,22 @@ export async function getJSONSchemaOptionsFromOpenAPIOptions(
             if (callbackUrlRefKey.startsWith('$')) {
               continue;
             }
-            const pubsubTopic = callbackUrlRefKey
+            const pubsubTopicSuffix = callbackUrlRefKey
               .split('$request.query')
               .join('args')
               .split('$request.body#/')
+              .join('args.')
+              .split('$response.body#/')
               .join('args.');
             const callbackOperationConfig: JSONSchemaPubSubOperationConfig = {
               type: OperationTypeNode.SUBSCRIPTION,
               field: '',
-              pubsubTopic,
+              pubsubTopic: '',
             };
             const callbackUrlObj = callbackObj[callbackUrlRefKey];
             for (const method in callbackUrlObj) {
               const callbackOperation: OpenAPIV3.OperationObject = callbackUrlObj[method];
+              callbackOperationConfig.pubsubTopic = `webhook:${method}:${pubsubTopicSuffix}`;
               callbackOperationConfig.field = callbackOperation.operationId;
               callbackOperationConfig.description = callbackOperation.description || callbackOperation.summary;
               const requestBodyContents = (callbackOperation.requestBody as OpenAPIV3.RequestBodyObject)?.content;

packages/loaders/openapi/tests/example_api7_server.ts

@@ -46,7 +46,7 @@ export function startServer(HTTP_PORT: number) {
     if (req.body.userName && req.body.name) {
       const device = req.body;
       Devices[device.name] = device;
-      pubsub.publish(`/api/${device.userName}/devices/${req.method.toUpperCase()}`, device);
+      pubsub.publish(`webhook:post:/api/${device.userName}/devices/${req.method.toUpperCase()}`, device);
       res.status(200).send(device);
     } else {
       res.status(404).send({
@@ -71,7 +71,7 @@ export function startServer(HTTP_PORT: number) {
         const device = req.body;
         delete Devices[req.params.deviceName];
         Devices[device.deviceName] = device;
-        pubsub.publish(`/api/${device.userName}/devices/${req.method.toUpperCase()}`, device);
+        pubsub.publish(`webhook:post:/api/${device.userName}/devices/${req.method.toUpperCase()}`, device);
         res.status(200).send(device);
       } else {
         res.status(404).send({