mongodb and mongoose conflicts.

[jaymaniya247]

Hello,

i am using repository's code for connect mongodb with my subscriptions. i recently foundout that mongodb and mongoose libraries are conflicting. That's why i am not getting any payload. I am getting my payload undefined each time.

i am using mongoose for other tasks and mongodb for only pub-sub. In this way it was conflicting. Then i refered your code, in this i am getting type error which is because of conflict in mongodb library.

[mjwheatley]

What versions are you using in your project for mongodb and mongoose?

This repository as well as the dependency graphql-mongodb-subscriptions are both using mongodb@5.3.0 and mongoose@7.1.0

Please try that combination if those are not the versions you are using and let me know if it is still not working for you.

[jaymaniya247]

Hello, thank you so much for Your help it means a lot.

Yes, i changed versions and it worked. But my actual issue is that i am not getting any payload in withFilter to add some conditions for one to one chat.

in short my payload is undefined when i use graphql-mongodb-subscriptions. Also it is working with the local pubsub provided by graphql-subscriptions.

My server setup. main file.

// standalone server testing
// import { typeDefs } from "./schema";
// import { resolvers } from "./resolvers";
// import { startStandaloneServer } from "@apollo/server/standalone";

import { Express, json } from "express";
import cors from "cors";
import http from "http";
import { ApolloServer } from "@apollo/server";
import { expressMiddleware } from "@apollo/server/express4";
import { NODE_ENV, PORT } from "../config/config";

// ** Plugins
import { ApolloServerPluginInlineTraceDisabled } from "@apollo/server/plugin/disabled";
import {
  ApolloServerPluginLandingPageLocalDefault,
  ApolloServerPluginLandingPageProductionDefault,
} from "@apollo/server/plugin/landingPage/default";

// ** merged schema
import schema from "./merge";
import { throwGQError } from "./errors";

//** for socket */
import { ApolloServerPluginDrainHttpServer } from "@apollo/server/plugin/drainHttpServer";
import { WebSocketServer } from "ws";
import { useServer } from "graphql-ws/lib/use/ws";
import { getPubSub, localPubSub } from "./pubsub/mongoPubSub";

const corsOptions = {
  origin: "*",
  credentials: false,
};

export async function startApolloServer(
  httpServer: http.Server<
    typeof http.IncomingMessage,
    typeof http.ServerResponse
  >,
  app: Express
) {
  // ** Web socket
  // Creating the WebSocket server
  const wsServer = new WebSocketServer({
    // This is the `httpServer` we created in a previous step.
    server: httpServer,
    // Pass a different path here if app.use
    // serves expressMiddleware at a different path
    path: "/graphql/ws",
  });

  // ** pubsub
  const pubsub = await getPubSub();

  // Hand in the schema we just created and have the
  // WebSocketServer start listening.
  const serverCleanup = useServer(
    { schema, context: () => ({ user: null, pubsub }) },
    wsServer
  ); // ! change from i-message clone, from skype

  const server = new ApolloServer({
    schema,
    plugins: [
      // Proper shutdown for the HTTP server.
      ApolloServerPluginDrainHttpServer({ httpServer }),
      // Proper shutdown for the WebSocket server.
      {
        async serverWillStart() {
          return {
            async drainServer() {
              await serverCleanup.dispose();
            },
          };
        },
      },
      // ApolloServerPluginInlineTraceDisabled(),
      // NODE_ENV === "production"
      //   ? ApolloServerPluginLandingPageProductionDefault()
      //   : ApolloServerPluginLandingPageLocalDefault({ embed: false }),
    ],
  });

  await server
    .start()
    .then(() => {
      console.log(
        `🚀 Graphql Server ready on http://localhost:${PORT}/graphql`
      );
    })
    .catch((error) => {
      console.log(error);
    });

  app.use(
    "/graphql",
    cors<cors.CorsRequest>(corsOptions),
    json(),
    expressMiddleware(server, {
      context: async ({ req }) => {
        const authHeader = req.headers.authorization || "";
        const token = authHeader.split(" ")[1];

        // ? No need for now
        // if (!token) {
        //   throwGQError({
        //     message: "No token provided",
        //     code: "UNAUTHORIZED",
        //   });
        // }

        try {
          const user = validateToken(token);
          if (!user) {
            // ! Do something when token is not there.
            // throwGQError({
            //   message: "Invalid token",
            //   code: "UNAUTHORIZED",
            // });
          }

          return { user, pubsub };
        } catch (error) {
          throwGQError({
            message: "Unauthorized: " + error.message,
            code: "UNAUTHORIZED",
          });
        }
      },
    })
  );

  // const { url } = await startStandaloneServer(server, {
  //   context: async ({ req }) => {
  //     const token = req.headers.authorization || "";
  //     // Validate the token here and return the user's context
  //     const user = validateToken(token);
  //     return { user };
  //   },
  // });
}

// standalone server
function validateToken(token: string) {
  if (!token) return null;
  // Implement your token validation logic here
  // If the token is valid, return the user context
  // If not, throw an error or return null

  return token;
}

merge schemas and resolvers.

// ** safe and gives errors
const schema = makeExecutableSchema({
  typeDefs: [
    BooksType,
    BlogsType,
    UserType,
    MessageType,
    GroupType,
    BroadcastType,
  ],
  resolvers: [
    BooksResolver,
    BlogsResolver,
    UserResolver,
    MessageResolver,
    GroupResolver,
    BroadcastResolver,
  ],
});

export default schema;

my subscription implementation.

  Mutation: {
    sendMessageTest: async (
      payload: any,
      { message }: { message: MessageTestInput },
      context: GraphQLContext
    ) => {
      const { pubsub } = context;

      const { user, content } = message;
      const sendMessageTest = { user, content, timestamp: new Date() };

      messages.push(sendMessageTest); // Assuming messages is declared and accessible here
      pubsub.publish("NEW_MESSAGE", { sendMessageTest });
      return sendMessageTest;
    },
  },

  // ------------------------------->  subscription code <---------------------------
  sendMessageTest: {
      subscribe: withFilter(
        (_: any, __: any, context: GraphQLContext) => {
          const { pubsub } = context;
          return pubsub.asyncIterator(["NEW_MESSAGE"]);
        },
        (payload: any, args) => {
          console.log("payload: ", payload, "args: ", args);
          // return true;
          if (!payload) return false;

          // Filter messages based on user if needed
          return payload.sendMessageTest?.user === args.user;
        }
      ),
    },

My pubsubs

import { MongoClient, Db, ServerApiVersion } from "mongodb";
import { MongodbPubSub } from "graphql-mongodb-subscriptions";
import { MONGO_URI } from "../../config/config";
import { PubSub } from "graphql-subscriptions";
import mongoose from "mongoose";

export type CommonMessageHandler = (message: any) => any;

export interface MongoPubSubChannelOptions {
  size: number;
  max: number;
}

export interface PubSubMongoDbOptions {
  connectionDb: Db;
  channelName?: string;
  channelOptions?: MongoPubSubChannelOptions;
  connectionListener?: (event: string, data: any) => void;
  commonMessageHandler?: CommonMessageHandler;
}

export const createMongoPubSub = (options: PubSubMongoDbOptions) => {};

// ** client for pubsub
const client = new MongoClient(MONGO_URI, {
  serverApi: {
    version: ServerApiVersion.v1,
    deprecationErrors: true,
  },
});

(async () => await client.connect())();
const db = client.db("pubsubdb");

export const getPubSub = async () => {
  const pubsub = new MongodbPubSub({
    connectionDb: mongoose.connections[0].db,
    // channelName: "mubsub",
    // channelOptions: {
    //   size: 100000,
    //   max: 100000,
    // },
    connectionListener: (event, data) => {
      console.log("Connection Listener: ", event, data);
    },
    commonMessageHandler: (message) => {
      console.log("Connection Handler Message is :", message);
    },
  });

  return pubsub;
};

export const localPubSub = new PubSub();

virsons i am using ( i changed this one ) NOW conflict error is gone but payload is undefined when i use graphql-mongodb-subscriptions.

 "mongodb": "^5.3.0",
 "mongoose": "^7.1.0",

1) i tried with both mongoose and mongodb. both did not work. not getting payload. 2) my local pub sub is working just fine.

Please help me with this one.

[mjwheatley]

Your subscription withFilter() looks the same as in this sample repo. Perhaps it an issue with your connection to the Db?

Try changing the following in your pubsubs file:

// ** client for pubsub
const client = new MongoClient(MONGO_URI, {
  serverApi: {
    version: ServerApiVersion.v1,
    deprecationErrors: true,
  },
});

(async () => await client.connect())();
const db = client.db("pubsubdb");

export const getPubSub = async () => {
  const pubsub = new MongodbPubSub({
    connectionDb: mongoose.connections[0].db,
    // channelName: "mubsub",
    // channelOptions: {
    //   size: 100000,
    //   max: 100000,
    // },
    connectionListener: (event, data) => {
      console.log("Connection Listener: ", event, data);
    },
    commonMessageHandler: (message) => {
      console.log("Connection Handler Message is :", message);
    },
  });

  return pubsub;
};

to this:

export const getPubSub = async () => {
  await mongoose.connect(MONGODB_URI);
  const pubsub = new MongodbPubSub({
    connectionDb: mongoose.connections[0].db
  });

  return pubsub;
}

[jaymaniya247]

Thank you for your help. It was not getting proper connection. Now it is working properly.

before i was passing connection which was from connection file. now i am making a new connection and using it. ( In this way, it worked ).

My working code.

export const getPubSub = async () => {
  // Help code
  await mongoose.connect(MONGO_URI);
  const pubsub = new MongodbPubSub({
    connectionDb: mongoose.connections[0].db,
  });

  return pubsub;
};