From fc4b952aaef7f18e22945210cbd757bda200a233 Mon Sep 17 00:00:00 2001 From: cyph3rasi Date: Wed, 15 Jul 2026 07:45:30 -0700 Subject: [PATCH] Fix E2EE chat request storm and make inbox loading local-only Hop-State: A_06FPCFBG3R2YRJWE6EWP2K0 Hop-Proposal: R_06FPCFA4W3TYWKN560DQTS8 Hop-Task: T_06FPCE0N17V3VENMZWMWWSR Hop-Attempt: AT_06FPCE0N14C36Q1AM8KYK28 --- .../swarm/chat/conversations/route.test.ts | 119 ++++++++++++++ src/app/api/swarm/chat/conversations/route.ts | 151 +++++++++--------- src/app/chat/page.tsx | 37 +++-- src/lib/e2ee/use-e2ee-identity.ts | 10 +- 4 files changed, 224 insertions(+), 93 deletions(-) create mode 100644 src/app/api/swarm/chat/conversations/route.test.ts diff --git a/src/app/api/swarm/chat/conversations/route.test.ts b/src/app/api/swarm/chat/conversations/route.test.ts new file mode 100644 index 0000000..fb20036 --- /dev/null +++ b/src/app/api/swarm/chat/conversations/route.test.ts @@ -0,0 +1,119 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +const mocks = vi.hoisted(() => ({ + findConversations: vi.fn(), + getSession: vi.fn(), + select: vi.fn(), +})); + +vi.mock('@/lib/auth', () => ({ getSession: mocks.getSession })); + +vi.mock('@/db', async () => { + const schema = await vi.importActual('@/db/schema'); + return { + ...schema, + db: { + query: { + chatConversations: { findMany: mocks.findConversations }, + }, + select: mocks.select, + }, + }; +}); + +import { GET } from './route'; + +function selectResult(rows: unknown[], grouped = false) { + const terminal = vi.fn().mockResolvedValue(rows); + const where = grouped + ? vi.fn(() => ({ groupBy: terminal })) + : terminal; + return { + from: vi.fn(() => ({ where })), + }; +} + +describe('GET /api/swarm/chat/conversations', () => { + beforeEach(() => { + vi.clearAllMocks(); + vi.stubEnv('NEXT_PUBLIC_NODE_DOMAIN', 'local.example'); + vi.stubGlobal('fetch', vi.fn()); + mocks.getSession.mockResolvedValue({ + user: { + id: 'owner-id', + did: 'did:key:owner', + handle: 'owner', + publicKey: 'owner-signing-key', + }, + }); + }); + + it('batches local metadata and never contacts remote nodes on the inbox path', async () => { + mocks.findConversations.mockResolvedValue([ + { + id: 'remote-conversation', + participant1Id: 'owner-id', + participant2Handle: 'alice@offline.example', + messages: [], + }, + { + id: 'local-conversation', + participant1Id: 'owner-id', + participant2Handle: 'bob@local.example', + messages: [], + }, + ]); + mocks.select + .mockReturnValueOnce(selectResult([ + { conversationId: 'remote-conversation', count: 2 }, + ], true)) + .mockReturnValueOnce(selectResult([ + { + handle: 'alice@offline.example', + displayName: 'Alice', + avatarUrl: null, + did: 'did:key:alice', + publicKey: 'alice-signing-key', + isBot: false, + }, + { + handle: 'bob', + displayName: 'Bob', + avatarUrl: '/bob.png', + did: 'did:key:bob', + publicKey: 'bob-signing-key', + isBot: false, + }, + ])); + + const response = await GET(); + const body = await response.json(); + + expect(response.status).toBe(200); + expect(mocks.select).toHaveBeenCalledTimes(2); + expect(fetch).not.toHaveBeenCalled(); + expect(body.conversations).toMatchObject([ + { + id: 'remote-conversation', + participant2: { handle: 'alice@offline.example', displayName: 'Alice' }, + unreadCount: 2, + }, + { + id: 'local-conversation', + participant2: { handle: 'bob', displayName: 'Bob' }, + unreadCount: 0, + }, + ]); + }); + + it('returns an empty inbox without issuing aggregate queries', async () => { + mocks.findConversations.mockResolvedValue([]); + + const response = await GET(); + + expect(response.status).toBe(200); + await expect(response.json()).resolves.toEqual({ conversations: [] }); + expect(mocks.select).not.toHaveBeenCalled(); + expect(fetch).not.toHaveBeenCalled(); + }); +}); diff --git a/src/app/api/swarm/chat/conversations/route.ts b/src/app/api/swarm/chat/conversations/route.ts index 47464e3..100fb4c 100644 --- a/src/app/api/swarm/chat/conversations/route.ts +++ b/src/app/api/swarm/chat/conversations/route.ts @@ -5,8 +5,8 @@ */ import { NextResponse } from 'next/server'; -import { db, chatMessages } from '@/db'; -import { eq, and, isNull, sql } from 'drizzle-orm'; +import { db, chatMessages, users } from '@/db'; +import { and, inArray, isNull, ne, sql } from 'drizzle-orm'; import { getSession } from '@/lib/auth'; import { E2EE_CHAT_ACTION, E2EE_PROTOCOL_VERSION, e2eeMessageEnvelopeSchema } from '@/lib/e2ee/protocol'; @@ -33,82 +33,80 @@ export async function GET() { }, }); - // Calculate unread count for each conversation - const conversationsWithUnread = await Promise.all( - conversations.map(async (conv) => { - const unreadCount = await db - .select({ count: sql`count(*)` }) + const conversationIds = conversations.map((conversation) => conversation.id); + const participantLookupHandles = new Set(); + const localNodeDomain = process.env.NEXT_PUBLIC_NODE_DOMAIN; + for (const conversation of conversations) { + const participantHandle = conversation.participant2Handle; + participantLookupHandles.add(participantHandle); + const separator = participantHandle.lastIndexOf('@'); + if (separator > 0 && participantHandle.slice(separator + 1) === localNodeDomain) { + participantLookupHandles.add(participantHandle.slice(0, separator)); + } + } + + // The inbox must be local-data-only. The previous implementation issued an + // unread query, a user query, and potentially an outbound federation request + // for every row. A single unavailable peer could therefore block the entire + // response. Batch local metadata here; independent federation paths refresh + // the remote-user cache without being on the inbox critical path. + const [unreadRows, cachedUsers] = await Promise.all([ + conversationIds.length > 0 + ? db + .select({ + conversationId: chatMessages.conversationId, + count: sql`count(*)`, + }) .from(chatMessages) - .where( - and( - eq(chatMessages.conversationId, conv.id), - isNull(chatMessages.readAt), - sql`${chatMessages.senderHandle} != ${session.user.handle}` - ) - ); + .where(and( + inArray(chatMessages.conversationId, conversationIds), + isNull(chatMessages.readAt), + ne(chatMessages.senderHandle, session.user.handle), + )) + .groupBy(chatMessages.conversationId) + : Promise.resolve([]), + participantLookupHandles.size > 0 + ? db + .select({ + handle: users.handle, + displayName: users.displayName, + avatarUrl: users.avatarUrl, + did: users.did, + publicKey: users.publicKey, + isBot: users.isBot, + }) + .from(users) + .where(inArray(users.handle, [...participantLookupHandles])) + : Promise.resolve([]), + ]); - // Parse participant info + const unreadByConversation = new Map( + unreadRows.map((row) => [row.conversationId, Number(row.count || 0)]), + ); + const usersByHandle = new Map(cachedUsers.map((user) => [user.handle, user])); + + const conversationsWithUnread = conversations.map((conv) => { const participant2Handle = conv.participant2Handle; - const isRemote = participant2Handle.includes('@'); - - let participant2Info = { - handle: participant2Handle, - displayName: participant2Handle, - avatarUrl: null as string | null, - did: '' as string, - }; - - // Try to get cached user info - let cachedUser = await db.query.users.findFirst({ - where: { handle: participant2Handle }, - }); - - // If not found, check if it's a local user with a domain suffix - if (!cachedUser && participant2Handle.includes('@')) { - const [handlePart, domainPart] = participant2Handle.split('@'); - if (!domainPart || domainPart === process.env.NEXT_PUBLIC_NODE_DOMAIN) { - cachedUser = await db.query.users.findFirst({ - where: { handle: handlePart }, - }); - } - } - - // LAZY LOAD: If remote and (not cached OR missing avatar), try to fetch it now - if (isRemote && (!cachedUser || !cachedUser.avatarUrl)) { - try { - const [rHandle, rDomain] = participant2Handle.split('@'); - const { fetchSwarmUserProfile } = await import('@/lib/swarm/interactions'); - const profileData = await fetchSwarmUserProfile(rHandle, rDomain, 0); - - if (profileData?.profile) { - const { upsertRemoteUser } = await import('@/lib/swarm/user-cache'); - await upsertRemoteUser({ - handle: participant2Handle, - displayName: profileData.profile.displayName, - avatarUrl: profileData.profile.avatarUrl || null, - did: profileData.profile.did || '', - isBot: profileData.profile.isBot || false, - publicKey: profileData.profile.publicKey, - }); - - // Re-query to get the new cached user - cachedUser = await db.query.users.findFirst({ - where: { handle: participant2Handle }, - }); + const separator = participant2Handle.lastIndexOf('@'); + const localHandle = separator > 0 + && participant2Handle.slice(separator + 1) === localNodeDomain + ? participant2Handle.slice(0, separator) + : null; + const cachedUser = usersByHandle.get(participant2Handle) + || (localHandle ? usersByHandle.get(localHandle) : undefined); + const participant2Info = cachedUser + ? { + handle: cachedUser.handle, + displayName: cachedUser.displayName || cachedUser.handle, + avatarUrl: cachedUser.avatarUrl || null, + did: cachedUser.did || '', } - } catch (e) { - console.error(`[Lazy Load] Failed for ${participant2Handle}:`, e); - } - } - - if (cachedUser) { - participant2Info = { - handle: cachedUser.handle, - displayName: cachedUser.displayName || cachedUser.handle, - avatarUrl: cachedUser.avatarUrl || null, - did: cachedUser.did || '', - }; - } + : { + handle: participant2Handle, + displayName: participant2Handle, + avatarUrl: null as string | null, + did: '', + }; const latest = conv.messages[0] || null; let lastMessage: { @@ -164,10 +162,9 @@ export async function GET() { isBot: cachedUser?.isBot || false, }, lastMessage, - unreadCount: Number(unreadCount[0]?.count || 0), + unreadCount: unreadByConversation.get(conv.id) || 0, }; - }) - ); + }); return NextResponse.json({ conversations: conversationsWithUnread.filter(c => !c.participant2.isBot), diff --git a/src/app/chat/page.tsx b/src/app/chat/page.tsx index 4938f35..741df2a 100644 --- a/src/app/chat/page.tsx +++ b/src/app/chat/page.tsx @@ -17,7 +17,7 @@ import { type StoredChatMessage, } from '@/lib/e2ee/client'; import { encryptE2EEMessage } from '@/lib/e2ee/client-crypto'; -import type { E2EEKeyBundle, E2EEMessageEnvelope } from '@/lib/e2ee/protocol'; +import type { E2EEKeyBundle, E2EEKeyMaterial, E2EEMessageEnvelope } from '@/lib/e2ee/protocol'; import { useE2EEIdentity } from '@/lib/e2ee/use-e2ee-identity'; interface Conversation { @@ -132,6 +132,7 @@ export default function ChatPage() { const appliedSharedPostRef = useRef(null); const messagesRequestRef = useRef(0); const conversationsRequestRef = useRef(0); + const conversationsAbortRef = useRef(null); const peerResolutionRef = useRef(0); const composeRequestRef = useRef(0); const sendRequestRef = useRef(0); @@ -141,8 +142,12 @@ export default function ChatPage() { const selectedConversationKeyRef = useRef(null); const preparedSendsRef = useRef(new Map()); const activeSendKeysRef = useRef(new Set()); + const e2eeMaterialRef = useRef<{ accountDid: string; material: E2EEKeyMaterial } | null>(null); renderedAccountDidRef.current = user?.did ?? null; + e2eeMaterialRef.current = user?.did && e2eeIdentity.state.status === 'ready' + ? { accountDid: user.did, material: e2eeIdentity.state.material } + : null; selectedConversationRef.current = selectedConversation; selectedConversationKeyRef.current = selectedConversation ? encryptionConversationKey(selectedConversation) @@ -201,8 +206,11 @@ export default function ChatPage() { const loadConversations = useCallback(async (isInitialLoad = true) => { const requestId = ++conversationsRequestRef.current; - const requestAccountDid = user?.did ?? null; + const requestAccountDid = renderedAccountDidRef.current; + const e2eeMaterial = e2eeMaterialRef.current; + conversationsAbortRef.current?.abort(); const controller = new AbortController(); + conversationsAbortRef.current = controller; const timeout = window.setTimeout(() => controller.abort(), CHAT_REQUEST_TIMEOUT_MS); try { if (isInitialLoad) { @@ -225,15 +233,14 @@ export default function ChatPage() { setLoading(false); } - if (user?.did && e2eeIdentity.state.status === 'ready') { - const material = e2eeIdentity.state.material; + if (requestAccountDid && e2eeMaterial?.accountDid === requestAccountDid) { nextConversations = await Promise.all(nextConversations.map(async (conversation) => { if (!conversation.lastMessage) return conversation; try { const { content } = await decryptStoredChatMessage( conversation.lastMessage, - user.did!, - material, + requestAccountDid, + e2eeMaterial.material, ); const normalized = content.replace(/\s+/g, ' ').trim(); const characters = Array.from(normalized); @@ -255,9 +262,10 @@ export default function ChatPage() { setConversationsError(null); } } catch (e) { - console.error("Failed to load conversations", e); - if (requestId === conversationsRequestRef.current - && renderedAccountDidRef.current === requestAccountDid) { + const isCurrentRequest = requestId === conversationsRequestRef.current + && renderedAccountDidRef.current === requestAccountDid; + if (isCurrentRequest) { + console.error("Failed to load conversations", e); setConversationsError(controller.signal.aborted ? 'The node took too long to load conversations. Try again.' : e instanceof TypeError @@ -266,12 +274,15 @@ export default function ChatPage() { } } finally { window.clearTimeout(timeout); + if (conversationsAbortRef.current === controller) { + conversationsAbortRef.current = null; + } if (requestId === conversationsRequestRef.current && renderedAccountDidRef.current === requestAccountDid) { setLoading(false); } } - }, [user?.did, e2eeIdentity.state]); + }, []); const markAsRead = useCallback(async (conversationId: string) => { const requestAccountDid = renderedAccountDidRef.current; @@ -641,7 +652,7 @@ export default function ChatPage() { // Load conversations useEffect(() => { - if (!user) return; + if (!user?.did || !activeE2EEKeyId) return; void loadConversations(true); const pollInterval = setInterval(() => { @@ -650,9 +661,11 @@ export default function ChatPage() { return () => { clearInterval(pollInterval); + conversationsAbortRef.current?.abort(); + conversationsAbortRef.current = null; conversationsRequestRef.current += 1; }; - }, [user, activeE2EEKeyId, loadConversations]); + }, [user?.did, activeE2EEKeyId, loadConversations]); // Handle Compose Intent useEffect(() => { diff --git a/src/lib/e2ee/use-e2ee-identity.ts b/src/lib/e2ee/use-e2ee-identity.ts index d2c3522..d23b489 100644 --- a/src/lib/e2ee/use-e2ee-identity.ts +++ b/src/lib/e2ee/use-e2ee-identity.ts @@ -27,8 +27,10 @@ export type E2EEIdentityState = | { status: 'ready'; material: E2EEKeyMaterial; vault: ConfiguredVaultStatus } | { status: 'error'; message: string }; +const LOADING_IDENTITY_STATE: E2EEIdentityState = { status: 'loading' }; + export function useE2EEIdentity(did?: string | null, handle?: string | null) { - const [state, setState] = useState({ status: 'loading' }); + const [state, setState] = useState(LOADING_IDENTITY_STATE); const [stateOwnerDid, setStateOwnerDid] = useState(null); const [busy, setBusy] = useState(false); const [actionError, setActionError] = useState(null); @@ -38,13 +40,13 @@ export function useE2EEIdentity(did?: string | null, handle?: string | null) { const bootstrap = useCallback(async () => { const generation = ++generationRef.current; if (!did) { - setState({ status: 'loading' }); + setState(LOADING_IDENTITY_STATE); setStateOwnerDid(null); setBusy(false); setActionError(null); return; } - setState({ status: 'loading' }); + setState(LOADING_IDENTITY_STATE); setStateOwnerDid(null); setBusy(false); setActionError(null); @@ -190,7 +192,7 @@ export function useE2EEIdentity(did?: string | null, handle?: string | null) { const visibleState: E2EEIdentityState = did && stateOwnerDid === did ? state - : { status: 'loading' }; + : LOADING_IDENTITY_STATE; return { state: visibleState, busy, actionError, setup, unlock, reset, retry: bootstrap }; }