GnosysLabs/Synapsis
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix E2EE chat request storm and make inbox loading local-only

Browse Source 
Hop-State: A_06FPCFBG3R2YRJWE6EWP2K0
Hop-Proposal: R_06FPCFA4W3TYWKN560DQTS8
Hop-Task: T_06FPCE0N17V3VENMZWMWWSR
Hop-Attempt: AT_06FPCE0N14C36Q1AM8KYK28
This commit is contained in:
cyph3rasi
2026-07-15 07:45:30 -07:00
committed by Hop
parent 97869daed0
commit fc4b952aae
4 changed files with 224 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/app/api/swarm/chat/conversations/route.test.ts
+119
View File
@@ -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<typeof import('@/db/schema')>('@/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();
});
});
src/app/api/swarm/chat/conversations/route.ts
+68 -71
View File
@@ -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<number>`count(*)` })
const conversationIds = conversations.map((conversation) => conversation.id);
const participantLookupHandles = new Set<string>();
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<number>`count(*)`,
})
.from(chatMessages)
.where(
and(
eq(chatMessages.conversationId, conv.id),
.where(and(
inArray(chatMessages.conversationId, conversationIds),
isNull(chatMessages.readAt),
sql`${chatMessages.senderHandle} != ${session.user.handle}`
)
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([]),
]);
const unreadByConversation = new Map(
unreadRows.map((row) => [row.conversationId, Number(row.count || 0)]),
);
const usersByHandle = new Map(cachedUsers.map((user) => [user.handle, user]));
// Parse participant info
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 },
});
}
} catch (e) {
console.error(`[Lazy Load] Failed for ${participant2Handle}:`, e);
}
}
if (cachedUser) {
participant2Info = {
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 || '',
};
}
: {
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),
src/app/chat/page.tsx
+24 -11
View File
@@ -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<string | null>(null);
const messagesRequestRef = useRef(0);
const conversationsRequestRef = useRef(0);
const conversationsAbortRef = useRef<AbortController | null>(null);
const peerResolutionRef = useRef(0);
const composeRequestRef = useRef(0);
const sendRequestRef = useRef(0);
@@ -141,8 +142,12 @@ export default function ChatPage() {
const selectedConversationKeyRef = useRef<string | null>(null);
const preparedSendsRef = useRef(new Map<string, PreparedSend>());
const activeSendKeysRef = useRef(new Set<string>());
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) {
const isCurrentRequest = requestId === conversationsRequestRef.current
&& renderedAccountDidRef.current === requestAccountDid;
if (isCurrentRequest) {
console.error("Failed to load conversations", e);
if (requestId === conversationsRequestRef.current
&& renderedAccountDidRef.current === requestAccountDid) {
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(() => {
src/lib/e2ee/use-e2ee-identity.ts
+6 -4
View File
@@ -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<E2EEIdentityState>({ status: 'loading' });
const [state, setState] = useState<E2EEIdentityState>(LOADING_IDENTITY_STATE);
const [stateOwnerDid, setStateOwnerDid] = useState<string | null>(null);
const [busy, setBusy] = useState(false);
const [actionError, setActionError] = useState<string | null>(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 };
}