diff --git a/app/api/messages/[messageId]/read/route.ts b/app/api/messages/[messageId]/read/route.ts new file mode 100644 index 0000000..43b86ee --- /dev/null +++ b/app/api/messages/[messageId]/read/route.ts @@ -0,0 +1,195 @@ +import { createClient } from "@/lib/supabase/server" +import { type NextRequest, NextResponse } from "next/server" + +/** + * POST /api/messages/[messageId]/read + * Mark a message as read by the current user + */ +export async function POST( + request: NextRequest, + { params }: { params: Promise<{ messageId: string }> }, +) { + try { + const { messageId } = await params + const supabase = await createClient() + + const { + data: { user }, + } = await supabase.auth.getUser() + + if (!user) { + return NextResponse.json({ error: "Unauthorized" }, { status: 401 }) + } + + if (!messageId) { + return NextResponse.json({ error: "Message ID is required" }, { status: 400 }) + } + + // Verify message exists and user has access to the room + const { data: message, error: messageError } = await supabase + .from("messages") + .select("id, room_id") + .eq("id", messageId) + .single() + + if (messageError || !message) { + return NextResponse.json({ error: "Message not found" }, { status: 404 }) + } + + // Verify user is a member of the room + const { data: membership, error: memberError } = await supabase + .from("room_members") + .select("id, removed_at") + .eq("room_id", message.room_id) + .eq("user_id", user.id) + .single() + + if (memberError || !membership) { + return NextResponse.json( + { error: "You are not a member of this room" }, + { status: 403 }, + ) + } + + if (membership.removed_at) { + return NextResponse.json( + { error: "You have been removed from this room" }, + { status: 403 }, + ) + } + + // Insert or update the read receipt + const { data: readReceipt, error: insertError } = await supabase + .from("message_reads") + .upsert( + { + message_id: messageId, + user_id: user.id, + read_at: new Date().toISOString(), + }, + { onConflict: "message_id,user_id" }, + ) + .select() + + if (insertError) { + console.error("[read-receipt] Error marking message as read:", insertError) + throw insertError + } + + return NextResponse.json( + { + success: true, + readReceipt: readReceipt?.[0], + }, + { status: 200 }, + ) + } catch (error) { + console.error("[read-receipt] POST error:", error) + return NextResponse.json( + { error: "Failed to mark message as read" }, + { status: 500 }, + ) + } +} + +/** + * GET /api/messages/[messageId]/read + * Get all read receipts for a message (who has read it and when) + */ +export async function GET( + request: NextRequest, + { params }: { params: Promise<{ messageId: string }> }, +) { + try { + const { messageId } = await params + const supabase = await createClient() + + const { + data: { user }, + } = await supabase.auth.getUser() + + if (!user) { + return NextResponse.json({ error: "Unauthorized" }, { status: 401 }) + } + + if (!messageId) { + return NextResponse.json({ error: "Message ID is required" }, { status: 400 }) + } + + // Verify message exists and user has access to the room + const { data: message, error: messageError } = await supabase + .from("messages") + .select("id, room_id, user_id") + .eq("id", messageId) + .single() + + if (messageError || !message) { + return NextResponse.json({ error: "Message not found" }, { status: 404 }) + } + + // Verify user is a member of the room + const { data: membership, error: memberError } = await supabase + .from("room_members") + .select("id, removed_at") + .eq("room_id", message.room_id) + .eq("user_id", user.id) + .single() + + if (memberError || !membership) { + return NextResponse.json( + { error: "You are not a member of this room" }, + { status: 403 }, + ) + } + + if (membership.removed_at) { + return NextResponse.json( + { error: "You have been removed from this room" }, + { status: 403 }, + ) + } + + // Get all read receipts for this message with user info + const { data: readReceipts, error: readError } = await supabase + .from("message_reads") + .select( + ` + id, + user_id, + read_at, + profiles:user_id ( + id, + display_name, + avatar_url, + wallet_address + ) + `, + ) + .eq("message_id", messageId) + .order("read_at", { ascending: true }) + + if (readError) { + console.error("[read-receipt] Error fetching read receipts:", readError) + throw readError + } + + const readCount = readReceipts?.length || 0 + const senderRead = readReceipts?.some((r) => r.user_id === message.user_id) || false + + return NextResponse.json( + { + messageId, + readCount, + senderRead, + readReceipts: readReceipts || [], + }, + { status: 200 }, + ) + } catch (error) { + console.error("[read-receipt] GET error:", error) + return NextResponse.json( + { error: "Failed to fetch read receipts" }, + { status: 500 }, + ) + } +} diff --git a/components/MessageReadReceiptIndicator.tsx b/components/MessageReadReceiptIndicator.tsx new file mode 100644 index 0000000..8707d78 --- /dev/null +++ b/components/MessageReadReceiptIndicator.tsx @@ -0,0 +1,106 @@ +import React, { useState, useEffect } from 'react'; +import { useMessageReadReceipts, type ReadReceipt } from '@/src/hooks/useMessageReadReceipts'; + +export interface MessageReadReceiptIndicatorProps { + messageId: string; + roomId: string; + senderId: string; + currentUserId: string; + isOwn: boolean; + disabled?: boolean; +} + +export function MessageReadReceiptIndicator({ + messageId, + roomId, + senderId, + currentUserId, + isOwn, + disabled = false, +}: MessageReadReceiptIndicatorProps) { + const [showDetails, setShowDetails] = useState(false); + const { readStatus, fetchReadReceipts } = useMessageReadReceipts({ + roomId, + userId: currentUserId, + enabled: !disabled + }); + + const messageReadData = readStatus[messageId]; + + useEffect(() => { + // Only fetch for own messages (sender can see read receipts) + if (isOwn && !messageReadData && !disabled) { + fetchReadReceipts(messageId); + } + }, [messageId, isOwn, messageReadData, disabled, fetchReadReceipts]); + + if (!isOwn || disabled || !messageReadData) { + return null; + } + + const { readCount, readReceipts } = messageReadData; + + if (readCount === 0) { + return ( +
+ + + +
+ ); + } + + return ( +
+ + + {showDetails && readReceipts.length > 0 && ( +
+
Read by:
+
+ {readReceipts.map((receipt: ReadReceipt) => ( +
+ {receipt.avatar_url && ( + {receipt.displayName} + )} +
+
{receipt.displayName}
+
+ {new Date(receipt.readAt).toLocaleTimeString()} +
+
+
+ ))} +
+
+ )} +
+ ); +} diff --git a/lib/websocket/server.ts b/lib/websocket/server.ts index bf5de36..6ff1211 100644 --- a/lib/websocket/server.ts +++ b/lib/websocket/server.ts @@ -412,6 +412,38 @@ export function createWebSocketServer(port: number = 3001) { break } + case "message_read": { + const readRoomId = message.payload.roomId + const readMessageId = message.payload.messageId + const readUserId = connection.userId + + if (!readUserId) { + ws.send( + JSON.stringify({ + type: "error", + payload: { message: "Not authenticated" }, + timestamp: Date.now(), + }), + ) + break + } + + // Broadcast read receipt to the room + // The actual read receipt will be persisted to the database via the REST API + broadcastToRoom(readRoomId, { + type: "message_read_receipt", + payload: { + messageId: readMessageId, + userId: readUserId, + displayName: connection.user?.displayName, + readAt: Date.now(), + roomId: readRoomId, + }, + timestamp: Date.now(), + }) + break + } + case "typing": { const typingRoomId = message.payload.roomId diff --git a/scripts/017_message_read_receipts.sql b/scripts/017_message_read_receipts.sql new file mode 100644 index 0000000..99550ec --- /dev/null +++ b/scripts/017_message_read_receipts.sql @@ -0,0 +1,73 @@ +-- Migration: Message Read Receipts +-- Description: Tracks when users read messages with timestamps for real-time read receipt support +-- Date: 2026-06-25 + +-- Create message_reads table to track read timestamps per message per user +CREATE TABLE IF NOT EXISTS public.message_reads ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + message_id UUID NOT NULL REFERENCES public.messages(id) ON DELETE CASCADE, + user_id UUID NOT NULL REFERENCES auth.users(id) ON DELETE CASCADE, + read_at TIMESTAMPTZ NOT NULL DEFAULT timezone('utc', now()), + created_at TIMESTAMPTZ NOT NULL DEFAULT timezone('utc', now()), + UNIQUE(message_id, user_id) +); + +-- Indexes for efficient queries +CREATE INDEX IF NOT EXISTS idx_message_reads_message_id + ON public.message_reads(message_id); + +CREATE INDEX IF NOT EXISTS idx_message_reads_user_id + ON public.message_reads(user_id); + +CREATE INDEX IF NOT EXISTS idx_message_reads_read_at + ON public.message_reads(read_at DESC); + +-- Composite index for querying reads by room and user +CREATE INDEX IF NOT EXISTS idx_message_reads_message_user + ON public.message_reads(message_id, user_id); + +-- Enable RLS +ALTER TABLE public.message_reads ENABLE ROW LEVEL SECURITY; + +-- Users can view read receipts for messages in rooms they are members of +CREATE POLICY "Users can view read receipts for accessible messages" + ON public.message_reads FOR SELECT + USING ( + EXISTS ( + SELECT 1 FROM public.messages m + INNER JOIN public.room_members rm ON m.room_id = rm.room_id + WHERE m.id = message_id + AND rm.user_id = auth.uid() + AND rm.removed_at IS NULL + ) + ); + +-- Users can mark their own messages as read +CREATE POLICY "Users can mark messages as read by themselves" + ON public.message_reads FOR INSERT + WITH CHECK ( + auth.uid() = user_id + AND EXISTS ( + SELECT 1 FROM public.messages m + INNER JOIN public.room_members rm ON m.room_id = rm.room_id + WHERE m.id = message_id + AND rm.user_id = auth.uid() + AND rm.removed_at IS NULL + ) + ); + +-- Comments for documentation +COMMENT ON TABLE public.message_reads IS + 'Tracks when users read messages with timestamps for real-time read receipt support'; + +COMMENT ON COLUMN public.message_reads.message_id IS + 'Reference to the message that was read'; + +COMMENT ON COLUMN public.message_reads.user_id IS + 'User who read the message'; + +COMMENT ON COLUMN public.message_reads.read_at IS + 'Timestamp when the message was marked as read'; + +COMMENT ON COLUMN public.message_reads.created_at IS + 'Timestamp when the read receipt record was created'; diff --git a/src/hooks/useMessageReadReceipts.ts b/src/hooks/useMessageReadReceipts.ts new file mode 100644 index 0000000..85e65bb --- /dev/null +++ b/src/hooks/useMessageReadReceipts.ts @@ -0,0 +1,216 @@ +import { useState, useCallback, useEffect, useRef } from 'react'; + +export interface ReadReceipt { + userId: string; + displayName: string; + avatar_url?: string; + wallet_address?: string; + readAt: number; +} + +interface MessageReadStatus { + [messageId: string]: { + readCount: number; + readReceipts: ReadReceipt[]; + isLoading: boolean; + }; +} + +interface UseMessageReadReceiptsOptions { + roomId?: string; + userId?: string; + enabled?: boolean; +} + +export function useMessageReadReceipts(options: UseMessageReadReceiptsOptions = {}) { + const { roomId, userId, enabled = true } = options; + const [readStatus, setReadStatus] = useState({}); + const [isMarkingRead, setIsMarkingRead] = useState<{ [key: string]: boolean }>({}); + const readTimeoutsRef = useRef<{ [key: string]: NodeJS.Timeout }>({}); + + /** + * Mark a message as read + * Debounced to avoid excessive API calls + */ + const markMessageAsRead = useCallback( + async (messageId: string) => { + if (!enabled || !messageId) return; + + // Cancel any pending timeout for this message + if (readTimeoutsRef.current[messageId]) { + clearTimeout(readTimeoutsRef.current[messageId]); + } + + // Debounce the actual API call by 500ms + readTimeoutsRef.current[messageId] = setTimeout(async () => { + try { + setIsMarkingRead((prev) => ({ ...prev, [messageId]: true })); + + const response = await fetch(`/api/messages/${messageId}/read`, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + }, + }); + + if (!response.ok) { + console.error(`Failed to mark message ${messageId} as read`); + return; + } + + const data = await response.json(); + + // Emit a WebSocket event to broadcast the read receipt + if (window.__websocket && roomId) { + window.__websocket.send({ + type: 'message_read', + payload: { + messageId, + roomId, + }, + }); + } + + // Update local state + setReadStatus((prev) => ({ + ...prev, + [messageId]: { + ...prev[messageId], + readCount: (prev[messageId]?.readCount || 0) + 1, + }, + })); + } catch (error) { + console.error(`Error marking message ${messageId} as read:`, error); + } finally { + setIsMarkingRead((prev) => ({ ...prev, [messageId]: false })); + } + }, 500); + }, + [enabled, roomId], + ); + + /** + * Fetch read receipts for a specific message + */ + const fetchReadReceipts = useCallback( + async (messageId: string) => { + if (!enabled || !messageId) return; + + try { + setReadStatus((prev) => ({ + ...prev, + [messageId]: { ...(prev[messageId] || {}), isLoading: true } as any, + })); + + const response = await fetch(`/api/messages/${messageId}/read`); + + if (!response.ok) { + console.error(`Failed to fetch read receipts for message ${messageId}`); + return; + } + + const data = await response.json(); + + setReadStatus((prev) => ({ + ...prev, + [messageId]: { + readCount: data.readCount || 0, + readReceipts: (data.readReceipts || []).map((r: any) => ({ + userId: r.user_id, + displayName: r.profiles?.display_name || 'Anonymous', + avatar_url: r.profiles?.avatar_url, + wallet_address: r.profiles?.wallet_address, + readAt: new Date(r.read_at).getTime(), + })), + isLoading: false, + }, + })); + } catch (error) { + console.error(`Error fetching read receipts for message ${messageId}:`, error); + setReadStatus((prev) => ({ + ...prev, + [messageId]: { + ...(prev[messageId] || {}), + isLoading: false, + } as any, + })); + } + }, + [enabled], + ); + + /** + * Handle incoming read receipt from WebSocket + */ + const handleReadReceiptUpdate = useCallback( + (event: any) => { + if (event.type !== 'message_read_receipt') return; + + const { messageId, userId: readUserId, displayName, readAt, roomId: eventRoomId } = event.payload; + + // Only update if it's for a message in the current room + if (roomId && roomId !== eventRoomId) return; + + setReadStatus((prev) => { + const current = prev[messageId] || { readCount: 0, readReceipts: [], isLoading: false }; + + // Check if this read receipt already exists + const existingIndex = current.readReceipts.findIndex((r) => r.userId === readUserId); + + let updatedReceipts: ReadReceipt[]; + if (existingIndex >= 0) { + // Update existing read receipt + updatedReceipts = [...current.readReceipts]; + updatedReceipts[existingIndex] = { + ...updatedReceipts[existingIndex], + readAt, + }; + } else { + // Add new read receipt + updatedReceipts = [ + ...current.readReceipts, + { + userId: readUserId, + displayName, + readAt, + }, + ]; + } + + return { + ...prev, + [messageId]: { + ...current, + readCount: updatedReceipts.length, + readReceipts: updatedReceipts.sort((a, b) => a.readAt - b.readAt), + }, + }; + }); + }, + [roomId], + ); + + // Cleanup on unmount + useEffect(() => { + return () => { + Object.values(readTimeoutsRef.current).forEach((timeout) => clearTimeout(timeout)); + }; + }, []); + + return { + readStatus, + markMessageAsRead, + fetchReadReceipts, + handleReadReceiptUpdate, + isMarkingRead, + }; +} + +// Make WebSocket accessible globally for the hook +declare global { + interface Window { + __websocket?: { + send: (message: any) => void; + }; + } +} diff --git a/types/websocket.ts b/types/websocket.ts index 39e7fee..6919e4d 100644 --- a/types/websocket.ts +++ b/types/websocket.ts @@ -2,6 +2,7 @@ export type WebSocketServerEventType = | "message" | "message_status_update" + | "message_read_receipt" | "room_join" | "room_leave" | "user_typing" @@ -21,6 +22,7 @@ export type WebSocketClientEventType = | "leave_room" | "send_message" | "message_delivered" + | "message_read" | "typing" | "stop_typing" | "wallet_event"