songtianlun/happy-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: wrap handlers in try catch to avoid crashing the server on db issues

Browse Source
This commit is contained in:
Steve Korshakov 2025-07-26 13:31:15 -07:00
parent b656562e56
commit 93b48e6e62
1 changed files with 527 additions and 471 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

58
sources/app/api.ts
View File

@ -789,6 +789,7 @@ export async function startApi(): Promise<{ app: FastifyInstance; io: Server }>
});
socket.on('session-alive', async (data: any) => {
try {
const { sid, time, thinking } = data;
let t = time;
if (typeof t !== 'number') {
@ -828,9 +829,13 @@ export async function startApi(): Promise<{ app: FastifyInstance; io: Server }>
thinking
}
});
} catch (error) {
log({ module: 'websocket', level: 'error' }, `Error in session-alive: ${error}`);
}
});
socket.on('session-end', async (data: any) => {
try {
const { sid, time } = data;
let t = time;
if (typeof t !== 'number') {
@ -870,9 +875,13 @@ export async function startApi(): Promise<{ app: FastifyInstance; io: Server }>
thinking: false
}
});
} catch (error) {
log({ module: 'websocket', level: 'error' }, `Error in session-end: ${error}`);
}
});
socket.on('message', async (data: any) => {
try {
const { sid, message, localId } = data;
log({ module: 'websocket' }, `Received message from socket ${socket.id}: sessionId=${sid}, messageLength=${message.length} bytes, connectionType=${connection.connectionType}, connectionSessionId=${connection.connectionType === 'session-scoped' ? connection.sessionId : 'N/A'}`);
@ -943,9 +952,13 @@ export async function startApi(): Promise<{ app: FastifyInstance; io: Server }>
},
skipSenderConnection: connection
});
} catch (error) {
log({ module: 'websocket', level: 'error' }, `Error in message handler: ${error}`);
}
});
socket.on('update-metadata', async (data: any, callback: (response: any) => void) => {
try {
const { sid, metadata, expectedVersion } = data;
// Validate input
@ -1007,10 +1020,16 @@ export async function startApi(): Promise<{ app: FastifyInstance; io: Server }>
// Send success response with new version via callback
callback({ result: 'success', version: expectedVersion + 1, metadata: metadata });
} catch (error) {
log({ module: 'websocket', level: 'error' }, `Error in update-metadata: ${error}`);
if (callback) {
callback({ result: 'error' });
}
}
});
socket.on('update-state', async (data: any, callback: (response: any) => void) => {
try {
const { sid, agentState, expectedVersion } = data;
// Validate input
@ -1078,10 +1097,17 @@ export async function startApi(): Promise<{ app: FastifyInstance; io: Server }>
// Send success response with new version via callback
callback({ result: 'success', version: expectedVersion + 1, agentState: agentState });
} catch (error) {
log({ module: 'websocket', level: 'error' }, `Error in update-state: ${error}`);
if (callback) {
callback({ result: 'error' });
}
}
});
// RPC register - Register this socket as a listener for an RPC method
socket.on('rpc-register', async (data: any) => {
try {
const { method } = data;
if (!method || typeof method !== 'string') {
@ -1108,10 +1134,15 @@ export async function startApi(): Promise<{ app: FastifyInstance; io: Server }>
socket.emit('rpc-registered', { method });
log({ module: 'websocket-rpc' }, `RPC method registered: ${method} on socket ${socket.id} (user: ${userId})`);
log({ module: 'websocket-rpc' }, `Active RPC methods for user ${userId}: ${Array.from(userRpcMap.keys()).join(', ')}`);
} catch (error) {
log({ module: 'websocket', level: 'error' }, `Error in rpc-register: ${error}`);
socket.emit('rpc-error', { type: 'register', error: 'Internal error' });
}
});
// RPC unregister - Remove this socket as a listener for an RPC method
socket.on('rpc-unregister', async (data: any) => {
try {
const { method } = data;
if (!method || typeof method !== 'string') {
@ -1135,10 +1166,15 @@ export async function startApi(): Promise<{ app: FastifyInstance; io: Server }>
}
socket.emit('rpc-unregistered', { method });
} catch (error) {
log({ module: 'websocket', level: 'error' }, `Error in rpc-unregister: ${error}`);
socket.emit('rpc-error', { type: 'unregister', error: 'Internal error' });
}
});
// RPC call - Call an RPC method on another socket of the same user
socket.on('rpc-call', async (data: any, callback: (response: any) => void) => {
try {
const { method, params } = data;
if (!method || typeof method !== 'string') {
@ -1223,14 +1259,28 @@ export async function startApi(): Promise<{ app: FastifyInstance; io: Server }>
});
}
}
} catch (error) {
log({ module: 'websocket', level: 'error' }, `Error in rpc-call: ${error}`);
if (callback) {
callback({
ok: false,
error: 'Internal error'
});
}
}
});
socket.on('ping', async (callback: (response: any) => void) => {
try {
callback({});
} catch (error) {
log({ module: 'websocket', level: 'error' }, `Error in ping: ${error}`);
}
});
// Usage reporting
socket.on('usage-report', async (data: any, callback?: (response: any) => void) => {
try {
const { key, sessionId, tokens, cost } = data;
// Validate required fields
@ -1342,6 +1392,12 @@ export async function startApi(): Promise<{ app: FastifyInstance; io: Server }>
callback({ success: false, error: 'Failed to save usage report' });
}
}
} catch (error) {
log({ module: 'websocket', level: 'error' }, `Error in usage-report handler: ${error}`);
if (callback) {
callback({ success: false, error: 'Internal error' });
}
}
});
socket.emit('auth', { success: true, user: userId });