WebSockets and Real-Time Apps 2026 — Socket.io, SSE, and Pusher Guide

npm install socket.io
npm install -D @types/socket.io

// src/socket/server.ts
import { Server } from 'socket.io'
import { createServer } from 'http'
import { app } from '../server'  // Fastify or Express instance
import { verifyToken } from '../lib/auth'
 
const httpServer = createServer(app.server ?? app)
const io = new Server(httpServer, {
  cors: { origin: process.env.CLIENT_URL, credentials: true },
  transports: ['websocket', 'polling'],
})
 
// Authentication middleware
io.use(async (socket, next) => {
  const token = socket.handshake.auth.token
  if (!token) return next(new Error('Authentication required'))
 
  const user = await verifyToken(token)
  if (!user) return next(new Error('Invalid token'))
 
  socket.data.user = user
  next()
})
 
// Room-based chat
io.on('connection', (socket) => {
  const userId = socket.data.user.id
  console.log(`User ${userId} connected`)
 
  socket.on('join:room', (roomId: string) => {
    socket.join(roomId)
    socket.to(roomId).emit('user:joined', { userId, roomId })
  })
 
  socket.on('message:send', async ({ roomId, content }: { roomId: string; content: string }) => {
    const message = await saveMessage({ roomId, userId, content })
 
    // Broadcast to everyone in the room including sender
    io.to(roomId).emit('message:new', message)
  })
 
  socket.on('typing:start', ({ roomId }: { roomId: string }) => {
    socket.to(roomId).emit('typing:update', { userId, isTyping: true })
  })
 
  socket.on('typing:stop', ({ roomId }: { roomId: string }) => {
    socket.to(roomId).emit('typing:update', { userId, isTyping: false })
  })
 
  socket.on('disconnect', (reason) => {
    console.log(`User ${userId} disconnected:`, reason)
    // Notify rooms the user was in
    socket.rooms.forEach(roomId => {
      io.to(roomId).emit('user:left', { userId })
    })
  })
})
 
httpServer.listen(4000)

// src/hooks/useSocket.ts
import { useEffect, useRef, useCallback } from 'react'
import { io, Socket } from 'socket.io-client'
 
export function useSocket(token: string) {
  const socketRef = useRef<Socket | null>(null)
 
  useEffect(() => {
    socketRef.current = io(process.env.NEXT_PUBLIC_SOCKET_URL!, {
      auth: { token },
      transports: ['websocket'],
    })
 
    return () => {
      socketRef.current?.disconnect()
    }
  }, [token])
 
  return socketRef
}
 
// src/components/ChatRoom.tsx
'use client'
import { useState, useEffect, useRef } from 'react'
import { useSocket } from '@/hooks/useSocket'
 
export function ChatRoom({ roomId, token }: { roomId: string; token: string }) {
  const [messages, setMessages] = useState<Message[]>([])
  const [typingUsers, setTypingUsers] = useState<string[]>([])
  const socketRef = useSocket(token)
  const bottomRef = useRef<HTMLDivElement>(null)
 
  useEffect(() => {
    const socket = socketRef.current
    if (!socket) return
 
    socket.emit('join:room', roomId)
 
    socket.on('message:new', (msg: Message) => {
      setMessages(prev => [...prev, msg])
      bottomRef.current?.scrollIntoView({ behavior: 'smooth' })
    })
 
    socket.on('typing:update', ({ userId, isTyping }: { userId: string; isTyping: boolean }) => {
      setTypingUsers(prev =>
        isTyping ? [...prev, userId] : prev.filter(id => id !== userId)
      )
    })
 
    return () => {
      socket.off('message:new')
      socket.off('typing:update')
    }
  }, [roomId, socketRef])
 
  function sendMessage(content: string) {
    socketRef.current?.emit('message:send', { roomId, content })
  }
 
  return (
    <div className="flex flex-col h-full">
      <div className="flex-1 overflow-y-auto p-4 space-y-2">
        {messages.map(msg => <MessageBubble key={msg.id} message={msg} />)}
        {typingUsers.length > 0 && (
          <p className="text-gray-500 text-sm">{typingUsers.join(', ')} is typing…</p>
        )}
        <div ref={bottomRef} />
      </div>
      <MessageInput onSend={sendMessage} socket={socketRef.current} roomId={roomId} />
    </div>
  )
}

// app/api/notifications/stream/route.ts (Next.js)
import { auth } from '@/auth'
 
export async function GET() {
  const session = await auth()
  if (!session) return new Response('Unauthorized', { status: 401 })
 
  const userId = session.user.id
  const encoder = new TextEncoder()
 
  const stream = new ReadableStream({
    start(controller) {
      function sendEvent(event: string, data: unknown) {
        controller.enqueue(
          encoder.encode(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`)
        )
      }
 
      sendEvent('connected', { message: 'Stream established' })
 
      // Subscribe to Redis pub/sub for this user
      const sub = subscribeToUserEvents(userId, (event) => {
        sendEvent(event.type, event.data)
      })
 
      // Heartbeat to keep connection alive
      const heartbeat = setInterval(() => {
        sendEvent('ping', { ts: Date.now() })
      }, 30_000)
 
      return () => {
        clearInterval(heartbeat)
        sub.unsubscribe()
      }
    },
  })
 
  return new Response(stream, {
    headers: {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache',
      Connection: 'keep-alive',
    },
  })
}

// React SSE client
'use client'
import { useEffect, useState } from 'react'
 
export function useNotifications() {
  const [notifications, setNotifications] = useState<Notification[]>([])
 
  useEffect(() => {
    const es = new EventSource('/api/notifications/stream')
 
    es.addEventListener('notification', (e) => {
      setNotifications(prev => [JSON.parse(e.data), ...prev])
    })
 
    es.onerror = () => {
      console.warn('SSE disconnected, will auto-reconnect')
    }
 
    return () => es.close()
  }, [])
 
  return notifications
}