Add Postgres adapter for socket communication

apps/api/src/main.ts

@@ -8,6 +8,7 @@ import { setupSwagger } from './config/swagger.config'
 import { setupExceptions } from './config/exceptions.config'
 import { setupValidation } from './config/validation.config'
 import { setupShutdownHooks } from './config/shutdown.config'
+import { PostgresIoAdapter } from './sockets/notifications/adapter'
 
 const globalPrefix = process.env.GLOBAL_PREFIX ?? 'api/v1'
 const logLevels: LogLevel[] = ['error', 'warn']
@@ -27,6 +28,10 @@ async function bootstrap() {
   app.setGlobalPrefix(globalPrefix)
   app.enableVersioning({ type: VersioningType.URI })
 
+  const postgresIoAdapter = new PostgresIoAdapter(app)
+  await postgresIoAdapter.connectToPostgres()
+  app.useWebSocketAdapter(postgresIoAdapter)
+
   const appVersion = process.env.APP_VERSION || 'unknown'
   setupHelmet(app)
   setupCors(app)

apps/api/src/sockets/notifications/adapter.ts

@@ -0,0 +1,45 @@
+import { Logger } from '@nestjs/common'
+import { IoAdapter } from '@nestjs/platform-socket.io'
+import { createAdapter } from '@socket.io/postgres-adapter'
+import { Pool } from 'pg'
+import { ServerOptions } from 'socket.io'
+
+const { DB_PORT, DB_USER, DB_PASS, DB_HOST } = process.env
+
+export class PostgresIoAdapter extends IoAdapter {
+  private adapterConstructor: ReturnType<typeof createAdapter>
+
+  async connectToPostgres(): Promise<void> {
+    const pool = new Pool({
+      user: DB_USER,
+      host: DB_HOST,
+      database: 'postgres',
+      password: DB_PASS,
+      port: Number(DB_PORT) || 5432,
+    })
+
+    await pool.connect()
+
+    await pool
+      .query(
+        `
+  CREATE TABLE IF NOT EXISTS socket_io_attachments (
+      id          bigserial UNIQUE,
+      created_at  timestamptz DEFAULT NOW(),
+      payload     bytea
+  );
+`,
+      )
+      .then(() => {
+        Logger.log('Table socket_io_attachments is ready')
+      })
+
+    this.adapterConstructor = createAdapter(pool)
+  }
+
+  createIOServer(port: number, options?: ServerOptions): any {
+    const server = super.createIOServer(port, options)
+    server.adapter(this.adapterConstructor)
+    return server
+  }
+}

apps/api/src/sockets/notifications/gateway.ts

@@ -6,13 +6,14 @@ import {
   WebSocketServer,
 } from '@nestjs/websockets'
 import { Logger } from '@nestjs/common'
+import { Server } from 'socket.io'
 
 @WebSocketGateway({ namespace: '/api/v1', transport: 'websocket' })
 export class NotificationGateway
   implements OnGatewayConnection, OnGatewayInit, OnGatewayDisconnect
 {
   constructor() {}
-  @WebSocketServer() server
+  @WebSocketServer() server: Server
 
   afterInit() {
     Logger.log('Websocket server initiated and ready to receive connections')

package.json

@@ -44,6 +44,9 @@
     "@sendgrid/client": "7.6.2",
     "@sendgrid/mail": "7.6.0",
     "@sentry/node": "6.19.7",
+    "@socket.io/postgres-adapter": "^0.3.1",
+    "@socket.io/postgres-emitter": "^0.1.0",
+    "@types/pg": "^8.6.6",
     "aws-sdk": "2.1199.0",
     "class-transformer": "0.5.1",
     "class-validator": "0.14.0",
@@ -58,6 +61,7 @@
     "nest-keycloak-connect": "1.9.0",
     "reflect-metadata": "0.1.13",
     "rxjs": "7.5.5",
+    "socket.io": "latest",
     "stripe": "9.16.0",
     "swagger-ui-express": "4.5.0",
     "tslib": "2.4.0"
@@ -97,6 +101,7 @@
     "prettier": "2.6.0",
     "prisma": "4.3.1",
     "prisma-dbml-generator": "0.9.1",
+    "socket.io-client": "^4.6.0",
     "supertest": "^6.3.3",
     "ts-jest": "^29.0.5",
     "ts-node": "^10.9.1",