@@ -2,8 +2,38 @@ import { type Context } from "hono";
22import { scheduleMemoryExtraction , generateAIResponse , extractImageData } from "../utils/model.utils.js" ;
33import prisma from "../config/prisma.config.js" ;
44import { NotFoundError } from "../utils/appError.utils.js" ;
5- import { streamText } from "hono/streaming" ;
5+ import { streamText as honoStreamText } from "hono/streaming" ;
66import logger from "../utils/logger.utils.js" ;
7+ import { streamLoading , streamText , streamError } from "../utils/stream.utils.js" ;
8+ import { type UploadedFile } from "@app/shared/src/schemas/message.schema.js"
9+
10+ async function handleImageExtraction ( messageFiles : UploadedFile [ ] , stream : any ) : Promise < string > {
11+ if ( messageFiles ?. length === 0 ) return "" ;
12+
13+ try {
14+ await streamLoading ( stream , "Analyzing image..." ) ;
15+ const extractionPromises = messageFiles . map ( ( file : UploadedFile ) => {
16+ if ( file . fileUrl ) {
17+ return extractImageData ( file . fileUrl ) ;
18+ }
19+ return Promise . resolve ( null ) ;
20+ } ) ;
21+
22+ const extractionResults = await Promise . allSettled ( extractionPromises ) ;
23+
24+ const successfulData = extractionResults
25+ . filter ( ( result ) => result . status === "fulfilled" && result . value )
26+ . map ( ( result : any ) => result . value as string ) ;
27+
28+ if ( successfulData . length > 0 ) {
29+ return `\n\nExtracted information from attached images:\n${ successfulData . join ( "\n---\n" ) } \n\nUse this information if relevant when answering.` ;
30+ }
31+ } catch ( error ) {
32+ logger . error ( { message : "Failed to process attached images" , error } ) ;
33+ }
34+
35+ return "" ;
36+ }
737
838export async function handleUserMessageResponse ( c : Context ) {
939 const requestStartTime = new Date ( ) ;
@@ -12,33 +42,11 @@ export async function handleUserMessageResponse(c: Context) {
1242 const { query, model, messageFiles } = c . get ( "body" ) ;
1343 const timezone = c . req . header ( "x-client-timezone" ) || "UTC" ;
1444
15- return streamText ( c , async ( stream ) => {
45+ return honoStreamText ( c , async ( stream ) => {
1646 let fullResponse = "" ;
1747 let finalQuery = query ;
1848
19- try {
20- if ( messageFiles && Array . isArray ( messageFiles ) && messageFiles . length > 0 ) {
21- await stream . write ( JSON . stringify ( { type : "loading" , content : "Analyzing image..." } ) + "\n" ) ;
22- const extractionPromises = messageFiles . map ( ( file : any ) => {
23- if ( file . fileUrl ) {
24- return extractImageData ( file . fileUrl ) ;
25- }
26- return Promise . resolve ( null ) ;
27- } ) ;
28-
29- const extractionResults = await Promise . allSettled ( extractionPromises ) ;
30-
31- const successfulData = extractionResults
32- . filter ( ( result ) => result . status === "fulfilled" && result . value )
33- . map ( ( result : any ) => result . value as string ) ;
34-
35- if ( successfulData . length > 0 ) {
36- finalQuery += `\n\nExtracted information from attached images:\n${ successfulData . join ( "\n---\n" ) } \n\nUse this information if relevant when answering.` ;
37- }
38- }
39- } catch ( error ) {
40- logger . error ( { message : "Failed to process attached images" , error } ) ;
41- }
49+ finalQuery += await handleImageExtraction ( messageFiles , stream ) ;
4250
4351 try {
4452 const [ chat , preferences , memories ] = await Promise . all ( [
@@ -48,7 +56,7 @@ export async function handleUserMessageResponse(c: Context) {
4856 ] ) ;
4957
5058 if ( ! chat ) {
51- await stream . write ( JSON . stringify ( { type : "error" , content : "Chat not found or unauthorized" } ) + "\n ") ;
59+ await streamError ( stream , "Chat not found or unauthorized" ) ;
5260 return ;
5361 }
5462
@@ -64,15 +72,15 @@ export async function handleUserMessageResponse(c: Context) {
6472 let isFirstChunk = true ;
6573 for await ( const chunk of aiStream ) {
6674 if ( isFirstChunk ) {
67- await stream . write ( JSON . stringify ( { type : "loading" , content : null } ) + "\n" ) ;
75+ await streamLoading ( stream , null ) ;
6876 isFirstChunk = false ;
6977 }
7078 fullResponse += chunk ;
71- await stream . write ( JSON . stringify ( { type : "text" , content : chunk } ) + "\n" ) ;
79+ await streamText ( stream , chunk ) ;
7280 }
7381
7482 if ( isFirstChunk ) {
75- await stream . write ( JSON . stringify ( { type : "loading" , content : null } ) + "\n" ) ;
83+ await streamLoading ( stream , null ) ;
7684 }
7785
7886 const filesToCreate = messageFiles ?. length ? messageFiles : undefined ;
@@ -84,7 +92,7 @@ export async function handleUserMessageResponse(c: Context) {
8492 messages : {
8593 create : [
8694 {
87- text : finalQuery ,
95+ text : query ,
8896 role : "USER" ,
8997 createdAt : requestStartTime ,
9098 messageFiles : filesToCreate ? { create : filesToCreate } : undefined ,
@@ -102,7 +110,7 @@ export async function handleUserMessageResponse(c: Context) {
102110 chatId,
103111 error : error . message ,
104112 } ) ;
105- await stream . write ( JSON . stringify ( { type : "error" , content : "Response generation interrupted" } ) + "\n ") ;
113+ await streamError ( stream , "Response generation interrupted" ) ;
106114 }
107115 } ) ;
108116}
0 commit comments