title: "Group messaging" in a MERN app using socket.io and mongodb change streams
body: <p><strong>Background:</strong></p>
  
  <p>Basically, there are registered users; these users correlate to documents in my mongodb cluster. These users have "institutions" ID's as properties in the documents, that "group" them. So for example, users will have institution ID's A, B, C, D; therefore there'll be groups A, B, C, D. These users post, update, and delete "post" documents that also have institution ID's of A, B, C, D. Group A members can only post, edit, or delete Group A posts, and the same for groups B, C, D.</p>
  
  <p>The current problem is communicating these changes in the backend to the frontend. I have a reactjs, redux front end, and an express, node, moongodb backend. I'm using socket.io to emit the insert, delete, update events I "catch" using mongodb change streams, to the react front end that listens for these emits with socket.io client. However, my strategy of putting the logged in users in rooms corresponding to their group ID's (so room A, room B, etc), and emitting the changes from the database to those rooms so each user only see the posts in their group, isn't working properly. What happens is that all users are notified of all database changes, in all groups. On the frontend, I end up getting group B posts, on a group A timeline, etc.</p>
  
  <p>I've attempted several solutions, and nothing seems to work; any help will be greatly appreciated.</p>
  
  <p><strong>Implementation</strong></p>
  
  <p><strong>backend:</strong></p>
  
  <pre><code>const express = require("express");
  const session = require('express-session');
  const mongoose = require("mongoose");
  const dotenv = require("dotenv").config();
  const bodyParser = require('body-parser');
  var cookieParser = require('cookie-parser')
  
  const passport = require('passport');
  
  const cors = require('cors');
  
  const { CLIENT_ORIGIN, PORT } = require('./config/prodConfig');
  
  const Post = require('./models/Post');
  
  const socket = require('socket.io');
  
  const events = require('events');
  
  
  const app = express();
  
  app.use(cors({
    origin: CLIENT_ORIGIN
  }));
  
  //body parser middleware
  app.use(express.json());
  app.use(cookieParser());
  
  // app.use(express.urlencoded({extended: true}));
  app.use( bodyParser.urlencoded({extended: true}));
  
  // Passport Config
  require('./config/passport')(passport);
  
  //Database config
  const db = process.env.MONGODB_URI;
  // console.log(db);
  
  //connect to MongoDB database
  mongoose
  .connect(db, {
      useNewUrlParser: true,
      useFindAndModify: false,
      useUnifiedTopology: true
      // useCreateIndex: true
  })
  .then(() =&gt; {
      console.log("Connected to MongoDB...");
  
      // change streams go here
      // const changeStream = Users.collection.watch();
      // changeStream.on('change', next =&gt; {
      //     console.log(next);
      // })
  })
  .catch(err =&gt; {
      console.log(err);
  });
  
  // Express session
  app.use(
      session({
        secret: process.env.EXPRESS_SESSION_SECRET+'',
        resave: true,
        saveUninitialized: true
      })
  );
  
  // Passport middleware
  app.use(passport.initialize());
  app.use(passport.session());
  
  
  //Server routes
  //add server route that allows "home landing path"
  app.use('/api/posts', require('./routes/api/posts'));
  app.use('/api/tags', require('./routes/api/tags'));
  app.use('/api/comments', require('./routes/api/comments'));
  app.use('/api/users', require('./routes/api/users'));
  app.use('/api/reactions', require('./routes/api/reactions'));
  app.use('/', require('./routes/landing'));
  
  
  
  //Serve static folder if in production
  
  if (process.env.NODE_ENV === 'production') {
      // Set static folder
      app.use(express.static('client/build'));
      app.enable('trust proxy');
  
      // Catch all to handle all other requests that come into the app. 
      app.get('*', (req, res) =&gt; {
        res.sendFile(path.resolve(__dirname, 'client', 'build', 'index.html'));
      });
    }
  
  
  
  
  
  const server = app.listen(PORT, () =&gt; {
      console.log("Server is running on port: ", PORT)
  });
  
  
  const io = socket.listen(server);
  
  
  io.origins(CLIENT_ORIGIN);
  
  let eventEmitter = new events.EventEmitter();
  let schId = "";
  
  
  io.sockets.on('connection', (socket) =&gt; {
    console.log("Connection attained");
  
    //listen for changes/inserts/deletes in all posts
  
    eventEmitter.emit('commenceListen', socket);
  
  
  });
  
  eventEmitter.on('commenceListen', (socket) =&gt; {
    setImmediate(() =&gt; {
      socket.on('listenForPosts', (schoolId) =&gt; {
        socket.join(schoolId, () =&gt; {
          console.log('listening for posts...');
          let rooms = Object.keys(socket.rooms);
          console.log("socket id: ", socket.id);
          console.log("rooms: ", rooms);
          const defaultRoom = rooms[0];
  
          schId = schoolId;
          eventEmitter.emit('watchDB', schoolId, socket);
  
        });
      });
  
      socket.on('stopListenForPosts', () =&gt; {
        changeStream.close();
        setTimeout(() =&gt; {
          socket.disconnect(true);
        }, 3000)
      });
    })
  })
  
  
  eventEmitter.on('watchDB', (schoolId, socket) =&gt; {
    console.log('watching db...');
    console.log(schoolId);
    setImmediate(() =&gt; {
      const pipeline = { $match: { 'fullDocument.institution': schoolId } };
      // const changeStream = Post.watch(pipeline);
      const changeStream = mongoose.connection.collection("posts").watch(pipeline);
      // console.log(changeStream);
      // console.log(mongoose.connection.collection("posts"));
  
      changeStream.on('change', next =&gt; {
        console.log('change stream on');
        // console.log(next);
        switch(next.operationType) {
          case 'insert':
            console.log('an insert happened...', "schoolId: ", next.fullDocument.institution);
            let rooms = Object.keys(socket.rooms);
            console.log("room to emit to: ", schoolId);
            console.log("rooms: ", rooms);
  
            socket.to(schoolId).emit('insert', {
              type: 'insert',
              msg: 'New question available',
              newPost: next.fullDocument
            });
            break;
  
          case 'update':
            console.log('an update happened...');
  
            socket.to(schoolId).emit('update', {
              type: 'update',
              postId: next.documentKey._id,
              updateInfo: next.updateDescription.updatedFields,
              msg: "Question has been updated."
            });
            break;
  
          case 'delete':
            console.log('a delete happened...');
  
            socket.to(schoolId).emit('delete', {
              type: 'delete',
              deletedId: next.documentKey._id,
              msg: 'Question has been deleted.'
            });
            break;
  
          default:
            break;
        }
      });
    });
  });
  </code></pre>
  
  <p><strong>frontend:</strong></p>
  
  <pre><code>import io from 'socket.io-client';
  const socket = io();
   socket.on('insert', (data) =&gt; {
       this.props.actions.postActions.doSocketInsert(data);
   });
  
   socket.on('update', (data) =&gt; {
       this.props.actions.postActions.doSocketUpdate(data);
   });
  
   socket.on('delete', (data) =&gt; {
       this.props.actions.postActions.doSocketDelete(data);
   });
  
   socket.on('connect', () =&gt; {
       this.listen(socket);
   });
  
   socket.on('restartListen', () =&gt; {
       socket.connect();
       this.listen(socket);
   });
  </code></pre>