feat: update a websocket live when changes are committed

src/controllers/staging.controller.js

@@ -87,6 +87,8 @@ export const commit = async (req, res) => {
           fullNode.deleteProjectRecord(uuid, stagingRecordId);
           break;
       }
+
+      Project.changes.next(data.orgUid);
     } else if (table === 'Unit') {
       switch (action) {
         case 'INSERT':
@@ -99,6 +101,7 @@ export const commit = async (req, res) => {
           fullNode.deleteUnitRecord(uuid, stagingRecordId);
           break;
       }
+      Unit.changes.next(data.orgUid);
     }
   });
   res.json({ message: 'Staging Table committed to full node' });

src/models/projects/projects.model.js

@@ -1,5 +1,7 @@
 'use strict';
 import Sequelize from 'sequelize';
+import rxjs from "rxjs";
+
 const { Model } = Sequelize;
 
 import { sequelize } from '../database';
@@ -15,6 +17,8 @@ import ModelTypes from './projects.modeltypes.cjs';
 import { optionallyPaginatedResponse } from "../../controllers/helpers.js";
 
 class Project extends Model {
+  static changes = new rxjs.Subject();
+
   static associate() {
     Project.hasMany(RelatedProject);
     Project.hasMany(Vintage);

src/models/units/units.model.js

@@ -4,10 +4,13 @@ import { sequelize } from '../database';
 import { Qualification, Vintage } from '../../models';
 
 import ModelTypes from './units.modeltypes.cjs';
+import rxjs from "rxjs";
 
 const { Model } = Sequelize;
 
 class Unit extends Model {
+  static changes = new rxjs.Subject();
+
   static associate() {
     Unit.hasOne(Vintage);
 

src/server.js

@@ -3,16 +3,45 @@
 import rootRouter from './routes';
 import http from 'http';
 import Debug from 'debug';
+import { Server } from "socket.io";
+import {Project, Unit} from "./models/index.js";
 
 const debug = Debug('climate-warehouse:server');
 
 const port = 3030;
 const server = http.createServer(rootRouter);
 
-server.listen(port);
+const io = new Server(server);
+
+const socketSubscriptions = {};
+
+io.on("connection", (socket) => {
+  const { url } = socket.request;
+  let subject;
+
+  if (url.includes('project')) {
+    subject = Project.changes;
+  } else if (url.includes('unit')) {
+    subject = Unit.changes;
+  } else {
+    return socket.disconnect();
+  }
+
+  socket.on('disconnect', () => {
+    if (socketSubscriptions[socket.id]) {
+      socketSubscriptions[socket.id].unsubscribe();
+    }
+
+  });
+
+  socketSubscriptions[socket.id] = subject.subscribe(orgUid => socket.broadcast.emit({ orgUid }))
+});
+
 server.on('error', onError);
 server.on('listening', onListening);
 
+server.listen(port);
+
 function onError(error) {
   if (error.syscall !== 'listen') {
     throw error;