graphql-subscriptions

backend/package.json

@@ -21,15 +21,18 @@
     "@fastify/cors": "^8.4.1",
     "@fastify/websocket": "^8.3.0",
     "@graphql-tools/merge": "^9.0.1",
+    "@graphql-tools/schema": "^10.0.2",
     "@hocuspocus/extension-database": "^2.8.1",
     "@hocuspocus/server": "^2.8.1",
     "axios": "^1.6.0",
     "fastify": "^4.24.3",
     "fastify-graceful-shutdown": "^3.5.1",
     "fastify-socket.io": "^5.0.0",
     "graphql": "^16.8.1",
+    "graphql-postgres-subscriptions": "^1.0.5",
     "graphql-tag": "^2.12.6",
     "graphql-upload": "^16.0.2",
+    "graphql-ws": "^5.14.3",
     "lodash-es": "^4.17.21",
     "minio": "^7.1.3",
     "pg": "^8.11.3",
@@ -46,6 +49,7 @@
   },
   "devDependencies": {
     "@trivago/prettier-plugin-sort-imports": "^4.3.0",
+    "@types/ws": "^8.5.10",
     "chai": "^4.3.10",
     "mocha": "^10.2.0",
     "prettier": "^3.0.3",

backend/src/index.js

@@ -2,10 +2,12 @@ import { ApolloServer } from '@apollo/server';
 import { fastifyApolloDrainPlugin, fastifyApolloHandler } from '@as-integrations/fastify';
 import * as cors from '@fastify/cors';
 import { fastifyWebsocket } from '@fastify/websocket';
+import { makeExecutableSchema } from '@graphql-tools/schema';
 import axios from 'axios';
 import Fastify from 'fastify';
 import fastifyIO from 'fastify-socket.io';
 import processRequest from 'graphql-upload/processRequest.mjs';
+import { makeHandler } from 'graphql-ws/lib/use/@fastify/websocket';
 import { GraphQLError } from 'graphql/error/index.js';
 
 import { db } from './db/index.js';
@@ -25,6 +27,8 @@ import { minioClient } from './services/minio-client.js';
 import { socketInit } from './socket/index.js';
 import typeDefs from './type-defs.js';
 
+const schema = makeExecutableSchema({ typeDefs, resolvers });
+
 const fastify = Fastify({
   logger: ENABLE_FASTIFY_LOGGING,
   keepAliveTimeout: 61 * 1000,
@@ -114,8 +118,13 @@ const apollo = new ApolloServer({
 await apollo.start();
 
 const myContextFunction = async (request) => {
+  let token = request?.headers?.authorization;
+
+  if (request?.connectionParams?.headers?.Authorization) {
+    token = request?.connectionParams?.headers?.Authorization;
+  }
+
   // get the user token from the headers
-  const token = request.headers.authorization;
   let user = null;
 
   // Allow if introspection query only
@@ -131,7 +140,7 @@ const myContextFunction = async (request) => {
     try {
       // TODO: maybe we just call the url of the caller origin
       const { data } = await axios.get(`${FRONTEND_HOSTNAME}/api/verify-jwt`, {
-        headers: { Authorization: request.headers.authorization },
+        headers: { Authorization: token },
       });
 
       // TODO: We can inject from DB here the whitelist domains and emails in addition to ENV vars
@@ -208,6 +217,10 @@ const myContextFunction = async (request) => {
   };
 };
 
+fastify.register(async (fastify) => {
+  fastify.get('/graphql', { websocket: true }, makeHandler({ schema, context: myContextFunction }));
+});
+
 fastify.post(
   '/graphql',
   fastifyApolloHandler(apollo, {

backend/src/resolvers/Board/index.js

@@ -1,4 +1,11 @@
+import pubsub from '../../services/apollo-pg-pubsub.js';
+
 const resolvers = {
+  Subscription: {
+    boardUpdated: {
+      subscribe: () => pubsub.asyncIterator(['BOARD_UPDATED']),
+    },
+  },
   Query: {
     boards: (parent, args, { db }) => {
       // TODO: should we require a project id to show boards?
@@ -27,6 +34,10 @@ const resolvers = {
 
       await board.save();
 
+      pubsub.publish('BOARD_UPDATED', {
+        boardUpdated: board,
+      });
+
       // TODO: fix
       // emitBoardUpdatedEvent(io, board.toJSON());
 

backend/src/resolvers/Issue/index.js

@@ -1,6 +1,13 @@
 import { Op } from 'sequelize';
 
+import pubsub from '../../services/apollo-pg-pubsub.js';
+
 const resolvers = {
+  Subscription: {
+    issueCreated: {
+      subscribe: () => pubsub.asyncIterator(['ISSUE_CREATED']),
+    },
+  },
   Query: {
     issues: (parent, { input: { projectId, id, search, searchOperator } }, { db }) => {
       let whereOr = [];
@@ -175,7 +182,13 @@ const resolvers = {
 
       const issueStatus = await db.sequelize.models.IssueStatuses.findByPk(issueStatusId);
 
-      return { ...issue.toJSON(), status: issueStatus.toJSON() };
+      const returnData = { ...issue.toJSON(), status: issueStatus.toJSON() };
+
+      pubsub.publish('ISSUE_CREATED', {
+        issueCreated: returnData,
+      });
+
+      return returnData;
     },
     deleteIssue: async (parent, { input }, { db }) => {
       const { id } = input;

backend/src/services/apollo-pg-pubsub.js

@@ -0,0 +1,14 @@
+import { PostgresPubSub } from 'graphql-postgres-subscriptions';
+import pkg from 'pg';
+
+import { SQL_URI } from './config.js';
+
+const { Client } = pkg;
+
+const client = new Client({
+  connectionString: SQL_URI,
+});
+await client.connect();
+const pubsub = new PostgresPubSub({ client });
+
+export default pubsub;

backend/src/type-defs.js

@@ -377,6 +377,12 @@ const typeDefs = gql`
     issues(input: QueryIssueInput): [Issue]
     issue(input: QueryIssueInput): Issue
   }
+
+  # SUBSCRIPTIONS
+  type Subscription {
+    issueCreated: Issue
+    boardUpdated: Board
+  }
 `;
 
 export default typeDefs;