Fix realtime updates for sidebar EmailsList and thread Conversations

app/api/webhooks/inbound-email/route.ts

@@ -92,7 +92,7 @@ export async function POST(request: Request) {
     );
   }
 
-  const { error: emailError } = await supabase
+  const { data: emailData, error: emailError } = await supabase
     .from('email')
     .insert({
       organization_id: orgData.id,
@@ -103,20 +103,32 @@ export async function POST(request: Request) {
       cleaned_body: text,
       role: 'user',
     })
+    .select('created_at')
     .single();
 
-  if (thread.status === 'closed') {
-    await supabase
-      .from('thread')
-      .update({ status: 'open' })
-      .match({ id: thread.id });
-  }
-
   if (emailError) {
     return new Response(JSON.stringify({ error: 'Failed to create email' }), {
       status: 200,
     });
   }
 
+  const threadUpdate: { last_message_created_at: string; status?: 'open' } = {
+    last_message_created_at: emailData.created_at,
+  };
+  if (thread.status === 'closed') {
+    threadUpdate.status = 'open';
+  }
+  const { error: threadUpdateError } = await supabase
+    .from('thread')
+    .update(threadUpdate)
+    .match({ id: thread.id });
+
+  if (threadUpdateError) {
+    return new Response(
+      JSON.stringify({ error: 'Failed to update thread' }),
+      { status: 200 }
+    );
+  }
+
   return new Response(JSON.stringify({ ok: true }), { status: 200 });
 }

components/organization/conversation/Conversations.tsx

@@ -1,12 +1,14 @@
 'use client';
 
-import { useEffect, useRef } from 'react';
+import { useEffect, useRef, useState } from 'react';
 import dynamic from 'next/dynamic';
 import { useRouter } from 'next/navigation';
 import { sendEmail } from '@/actions/email';
 import { Tables } from '@/database.types';
+import { RealtimeChannel } from '@supabase/supabase-js';
 
 import { useTiptap } from '@/hooks/useTiptap';
+import { createBrowserClient } from '@/lib/supabase/client';
 import { Badge } from '@/components/ui/badge';
 import { Button } from '@/components/ui/button';
 
@@ -45,12 +47,14 @@ function confidenceVariant(score: number): 'default' | 'neutral' {
 
 export function Conversations({
   threadId,
-  conversations,
+  conversations: initialConversations,
   reply,
   status,
 }: Props) {
   const router = useRouter();
   const divRef = useRef<HTMLDivElement>(null);
+  const channelRef = useRef<RealtimeChannel | null>(null);
+  const [conversations, setConversations] = useState(initialConversations);
 
   const editor = useTiptap(
     conversations[conversations.length - 1].role === 'user'
@@ -61,7 +65,39 @@ export function Conversations({
   useEffect(() => {
     if (!divRef.current) return;
     divRef.current.scrollTop = divRef.current.scrollHeight;
-  }, []);
+  }, [conversations]);
+
+  useEffect(() => {
+    if (channelRef.current) return;
+
+    const supabase = createBrowserClient();
+    channelRef.current = supabase
+      .channel(`email:${threadId}`)
+      .on(
+        'postgres_changes',
+        {
+          event: 'INSERT',
+          schema: 'public',
+          table: 'email',
+          filter: `thread_id=eq.${threadId}`,
+        },
+        (payload) => {
+          const newEmail = payload.new as Tables<'email'>;
+          setConversations((prev) =>
+            prev.some((e) => e.id === newEmail.id) ? prev : [...prev, newEmail]
+          );
+        }
+      )
+      .subscribe();
+
+    return () => {
+      if (channelRef.current) {
+        const channel = channelRef.current;
+        channelRef.current = null;
+        channel.unsubscribe();
+      }
+    };
+  }, [threadId]);
 
   const handleSubmit = async (
     status: 'open' | 'closed' | undefined = undefined

components/organization/sidebar/EmailsList.tsx

@@ -32,16 +32,38 @@ type ThreadAction =
       data: Tables<'thread'>[];
       status: 'open' | 'closed';
     }
-  | { type: 'INSERT_THREAD'; thread: Tables<'thread'> };
+  | { type: 'INSERT_THREAD'; thread: Tables<'thread'> }
+  | { type: 'UPDATE_THREAD'; thread: Tables<'thread'> };
 
 function threadReducer(state: ThreadState, action: ThreadAction): ThreadState {
   switch (action.type) {
     case 'FETCH_START':
       return { ...state, isLoading: true };
     case 'FETCH_SUCCESS':
       return { data: action.data, status: action.status, isLoading: false };
-    case 'INSERT_THREAD':
+    case 'INSERT_THREAD': {
+      if (state.data.some((t) => t.id === action.thread.id)) return state;
       return { ...state, data: [action.thread, ...state.data] };
+    }
+    case 'UPDATE_THREAD': {
+      const existsInList = state.data.some((t) => t.id === action.thread.id);
+      if (action.thread.status !== state.status) {
+        if (!existsInList) return state;
+        return {
+          ...state,
+          data: state.data.filter((t) => t.id !== action.thread.id),
+        };
+      }
+      const updated = existsInList
+        ? state.data.map((t) => (t.id === action.thread.id ? action.thread : t))
+        : [action.thread, ...state.data];
+      updated.sort(
+        (a, b) =>
+          new Date(b.last_message_created_at).getTime() -
+          new Date(a.last_message_created_at).getTime()
+      );
+      return { ...state, data: updated };
+    }
   }
 }
 
@@ -103,6 +125,21 @@ export function EmailsList({ orgId, name, slug, inboundEmail }: Props) {
           });
         }
       )
+      .on(
+        'postgres_changes',
+        {
+          event: 'UPDATE',
+          schema: 'public',
+          table: 'thread',
+          filter: `organization_id=eq.${orgId}`,
+        },
+        (payload) => {
+          dispatch({
+            type: 'UPDATE_THREAD',
+            thread: payload.new as Tables<'thread'>,
+          });
+        }
+      )
       .subscribe();
 
     return () => {