How can I authenticate my users over websocket in apolloServer so that i can protect the subscriptions

I am building an instagram clone. It was going well but when i got to authenticating the subscription I became stuck. I searched online, but i couldn't find a thing. Basically what I am trying to do is check for a JWTToken before i allow the user to connect to the subscription and gain access to the live messages.

here is my Code:

server.js

const { createServer } = require('http');
const express = require('express');
const {ApolloServerPluginDrainHttpServer,ApolloServerPluginLandingPageLocalDefault, AuthenticationError,} = require("apollo-server-core");
const { makeExecutableSchema } = require('@graphql-tools/schema');
const { WebSocketServer } = require('ws');
const { useServer } = require('graphql-ws/lib/use/ws');
const { ApolloServer } = require('apollo-server-express');
const {sequelize} = require('./models')


const resolvers = require('./graphql/Resolvers')
const typeDefs = require('./graphql/typeDefs');


(async function() {
  const app = express();
  const httpServer = createServer(app);
const contextMiddleware = require('./util/contextMiddleware')

const schema = makeExecutableSchema({
  typeDefs,
  resolvers,
  });

const wsServer = new WebSocketServer({
  server: httpServer,
  path: '/graphql',
});


const serverCleanup = useServer({ schema }, wsServer);

const server = new ApolloServer({
    schema,
    context: contextMiddleware,
    subscription: {
      onConnect: (context) => {
        if(!context.user){
         throw new AuthenticationError('unauthenticated')
        }
       },
    },
    csrfPrevention: true,
    cache: "bounded",
    plugins: [
      ApolloServerPluginDrainHttpServer({ httpServer }),
        {
        async serverWillStart() {
          return {
            async drainServer() {
              await serverCleanup.dispose();
            },
          };
        },
      },
      ApolloServerPluginLandingPageLocalDefault({ embed: true }),
    ],
  });

  await server.start();
  server.applyMiddleware({ app })

  const PORT = process.env.PORT || 4000;

httpServer.listen(PORT,()=>{
  console.log(` Server ready at ${PORT}`);
  sequelize.authenticate()
  .then(()=>{
    console.log('database connected')
  }).catch((err)=>{
    console.log(err)
  })
})
})(typeDefs, resolvers)

messages.js --- important snippet

   Mutation: {
        sendMessage: async(parent, {to, content}, {user})=>{
            try{
                if(!user) throw new AuthenticationError('Ya dont exist')
                const recipient = await Users.findOne({where:{username:to}})
                if(!recipient){
                    throw new UserInputError('User not found')
                }else if(recipient.username === user.username){
                    throw new UserInputError('You can not message yourself')
                }
                if(content.trim() === '') throw new UserInputError('message can not be empty')

                const message = await Message.create({
                    from: user.username,
                    to,
                    content,
                })
                pubsub.publish('NEW_MESSAGE', {newMessage: message})
                return message
            }catch(err){
                console.log(err)
                throw err
            }
        }
    },
    Subscription: {
        newMessage:{
            subscribe: (_,__,{user}) => {
                if(!user) throw new AuthenticationError('unathenticated')
                pubsub.asyncIterator(['NEW_MESSAGE'])}
        }
    }

  };

I am passing my user Authentication through the context and it is held in a middleware which returns user object. Here I tried getting my user object from the context and checking if user exist then go through with the subscription, else then throw an error

contextMiddleware.js

const jwt = require('jsonwebtoken')
const {JWT_SECRET} = require('../config/env.json')


module.exports= context =>{
    if(context.req && context.req.headers.authorization){
        const token = context.req.headers.authorization.split('Bearer ')[1]
        jwt.verify(token, JWT_SECRET, (err, decodedToken) => {
            if(err){
                // throw new AuthenticationError('Unauthenticated')
            }
        context.user = decodedToken 


        })
    }
    return context
}