Backend Realtime

Real-time Features Skill

Implement live, collaborative features using WebSockets, Supabase Realtime, and Server-Sent Events.

CRITICAL: Check Existing First

Before implementing ANY real-time feature, verify:

1. Check for existing real-time setup:

cat package.json | grep -i "socket\|realtime\|pusher\|ably"
rg "supabase.*channel|useSubscription|WebSocket" --type ts --type tsx

1. Check for existing patterns:

rg "on\('INSERT'\|on\('UPDATE'\|subscribe\(" --type ts
ls -la src/hooks/use*Realtime* src/lib/realtime* 2>/dev/null

1. Check Supabase config:

rg "createClient|supabaseUrl" --type ts -l
cat .env* | grep -i SUPABASE

Why: Real-time connections are stateful. Don't create duplicate subscriptions.

Supabase Realtime

Subscribe to Database Changes

'use client'
import { useEffect, useState } from 'react'
import { createClient } from '@/lib/supabase/client'

export function useRealtimeMessages(roomId: string) {
 const [messages, setMessages] = useState([])
 const supabase = createClient()

 useEffect(() => {
 // Initial fetch
 const fetchMessages = async () => {
 const { data } = await supabase
 .from('messages')
 .select('*')
 .eq('room_id', roomId)
 .order('created_at', { ascending: true })

 if (data) setMessages(data)
 }
 fetchMessages()

 // Subscribe to changes
 const channel = supabase
 .channel(`room:${roomId}`)
 .on(
 'postgres_changes',
 {
 event: 'INSERT',
 schema: 'public',
 table: 'messages',
 filter: `room_id=eq.${roomId}`,
 },
 (payload) => {
 setMessages((prev) => [...prev, payload.new as Message])
 }
 )
 .on(
 'postgres_changes',
 {
 event: 'DELETE',
 schema: 'public',
 table: 'messages',
 filter: `room_id=eq.${roomId}`,
 },
 (payload) => {
 setMessages((prev) =>
 prev.filter((m) => m.id !== payload.old.id)
 )
 }
 )
 .subscribe()

 return () => {
 supabase.removeChannel(channel)
 }
 }, [roomId, supabase])

 return messages
}

Presence (Online Users)

'use client'
import { useEffect, useState } from 'react'
import { createClient } from '@/lib/supabase/client'
import type { RealtimePresenceState } from '@supabase/supabase-js'

interface UserPresence {
 id: string
 name: string
 avatar: string
 online_at: string
}

export function usePresence(roomId: string, currentUser: UserPresence) {
 const [onlineUsers, setOnlineUsers] = useState([])
 const supabase = createClient()

 useEffect(() => {
 const channel = supabase.channel(`presence:${roomId}`)

 channel
 .on('presence', { event: 'sync' }, () => {
 const state = channel.presenceState()
 const users = Object.values(state).flat()
 setOnlineUsers(users)
 })
 .on('presence', { event: 'join' }, ({ newPresences }) => {
 console.log('User joined:', newPresences)
 })
 .on('presence', { event: 'leave' }, ({ leftPresences }) => {
 console.log('User left:', leftPresences)
 })
 .subscribe(async (status) => {
 if (status === 'SUBSCRIBED') {
 await channel.track(currentUser)
 }
 })

 return () => {
 channel.untrack()
 supabase.removeChannel(channel)
 }
 }, [roomId, currentUser, supabase])

 return onlineUsers
}

Broadcast (Custom Events)

'use client'
import { useEffect, useCallback } from 'react'
import { createClient } from '@/lib/supabase/client'

export function useBroadcast(channelName: string) {
 const supabase = createClient()
 const channelRef = useRef | null>(null)

 useEffect(() => {
 channelRef.current = supabase.channel(channelName)

 channelRef.current
 .on('broadcast', { event: 'cursor-move' }, ({ payload }) => {
 // Handle cursor position updates from others
 console.log('Cursor moved:', payload)
 })
 .on('broadcast', { event: 'typing' }, ({ payload }) => {
 // Handle typing indicators
 console.log('User typing:', payload)
 })
 .subscribe()

 return () => {
 if (channelRef.current) {
 supabase.removeChannel(channelRef.current)
 }
 }
 }, [channelName, supabase])

 const broadcast = useCallback((event: string, payload: unknown) => {
 channelRef.current?.send({
 type: 'broadcast',
 event,
 payload,
 })
 }, [])

 return { broadcast }
}

// Usage: Collaborative cursors
function CollaborativeCanvas() {
 const { broadcast } = useBroadcast('canvas:123')

 const handleMouseMove = (e: React.MouseEvent) => {
 broadcast('cursor-move', {
 userId: currentUser.id,
 x: e.clientX,
 y: e.clientY,
 })
 }
}

TanStack Query + Real-time

import { useQuery, useQueryClient } from '@tanstack/react-query'
import { useEffect } from 'react'

export function useRealtimeQuery(
 queryKey: string[],
 queryFn: () => Promise,
 table: string,
 filter?: string
) {
 const queryClient = useQueryClient()
 const supabase = createClient()

 const query = useQuery({
 queryKey,
 queryFn,
 })

 useEffect(() => {
 const channel = supabase
 .channel(`${table}-changes`)
 .on(
 'postgres_changes',
 {
 event: '*',
 schema: 'public',
 table,
 filter,
 },
 () => {
 // Invalidate and refetch on any change
 queryClient.invalidateQueries({ queryKey })
 }
 )
 .subscribe()

 return () => {
 supabase.removeChannel(channel)
 }
 }, [table, filter, queryKey, queryClient, supabase])

 return query
}

// Usage
function ProductList() {
 const { data: products } = useRealtimeQuery(
 ['products'],
 () => fetchProducts(),
 'products'
 )
}

Optimistic Updates

'use client'
import { useOptimistic, useTransition } from 'react'
import { addMessage } from '@/app/actions'

function Chat({ initialMessages }: { initialMessages: Message[] }) {
 const [isPending, startTransition] = useTransition()
 const [optimisticMessages, addOptimisticMessage] = useOptimistic(
 initialMessages,
 (state, newMessage: Message) => [...state, newMessage]
 )

 async function handleSubmit(formData: FormData) {
 const content = formData.get('content') as string

 // Optimistic update - instant UI feedback
 const optimisticMessage: Message = {
 id: crypto.randomUUID(),
 content,
 created_at: new Date().toISOString(),
 user_id: currentUser.id,
 pending: true,
 }

 startTransition(async () => {
 addOptimisticMessage(optimisticMessage)
 await addMessage(content) // Server Action
 })
 }

 return (
 
 {optimisticMessages.map((msg) => (
 
 {msg.content}
 
 ))}
 
 
 Send
 
 
 )
}

Server-Sent Events (SSE)

// app/api/events/route.ts
export async function GET() {
 const encoder = new TextEncoder()

 const stream = new ReadableStream({
 async start(controller) {
 const send = (data: unknown) => {
 controller.enqueue(
 encoder.encode(`data: ${JSON.stringify(data)}\n\n`)
 )
 }

 // Send initial data
 send({ type: 'connected', timestamp: Date.now() })

 // Subscribe to changes (e.g., from database)
 const interval = setInterval(() => {
 send({ type: 'heartbeat', timestamp: Date.now() })
 }, 30000)

 // Cleanup on close
 // Note: In production, use proper cleanup mechanism
 },
 })

 return new Response(stream, {
 headers: {
 'Content-Type': 'text/event-stream',
 'Cache-Control': 'no-cache',
 'Connection': 'keep-alive',
 },
 })
}

// Client hook
function useSSE(url: string) {
 const [data, setData] = useState(null)

 useEffect(() => {
 const eventSource = new EventSource(url)

 eventSource.onmessage = (event) => {
 setData(JSON.parse(event.data))
 }

 eventSource.onerror = () => {
 eventSource.close()
 // Implement reconnection logic
 }

 return () => eventSource.close()
 }, [url])

 return data
}

Typing Indicators

function useTypingIndicator(channelName: string, userId: string) {
 const [typingUsers, setTypingUsers] = useState([])
 const supabase = createClient()
 const timeoutRef = useRef()

 useEffect(() => {
 const channel = supabase.channel(channelName)

 channel
 .on('broadcast', { event: 'typing' }, ({ payload }) => {
 if (payload.userId !== userId) {
 setTypingUsers((prev) =>
 prev.includes(payload.userId) ? prev : [...prev, payload.userId]
 )

 // Remove after 3 seconds of no activity
 setTimeout(() => {
 setTypingUsers((prev) =>
 prev.filter((id) => id !== payload.userId)
 )
 }, 3000)
 }
 })
 .subscribe()

 return () => {
 supabase.removeChannel(channel)
 }
 }, [channelName, userId, supabase])

 const sendTyping = useCallback(() => {
 clearTimeout(timeoutRef.current)

 supabase.channel(channelName).send({
 type: 'broadcast',
 event: 'typing',
 payload: { userId },
 })
 }, [channelName, userId, supabase])

 return { typingUsers, sendTyping }
}

Validation

After implementing real-time features:

1. Connection handling → Reconnects on disconnect, shows status
2. Error handling → Graceful degradation when real-time unavailable
3. Memory leaks → All subscriptions cleaned up on unmount
4. Duplicate subscriptions → No multiple listeners for same event
5. Optimistic updates → UI feels instant, handles conflicts
6. Mobile → Works on spotty connections, battery efficient

Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

• Author: kensaurus
• Source: kensaurus/cursor-kenji
• License: MIT
• Homepage: https://github.com/kensaurus/cursor-kenji

Install and usage instructions live in the source repository linked above.