fix: parser

Author: astandrik

src/services/api/viewer.ts

@@ -397,7 +397,6 @@ export class ViewerAPI extends BaseYdbAPI {
         // Create parser state and tracking in closure
         let parserState = {
             lastProcessedLength: 0,
-            buffer: '',
         };
 
         return this.get<string>(

src/services/parsers/parseMultipart.ts

@@ -1,136 +1,108 @@
 /* eslint-disable camelcase */
 import type {StreamingChunk} from '../../types/store/streaming';
 
+// interface Headers {
+//     contentType?: string;
+//     contentLength?: number;
+// }
+
 interface MultipartState {
     lastProcessedLength: number;
-    buffer: string;
 }
 
 interface MultipartResult {
     chunks: StreamingChunk[];
     state: MultipartState;
 }
 
-function isCompleteChunk(chunk: string): boolean {
-    // Normalize all possible line ending combinations to \n
-    const normalizedChunk = chunk.replace(/\r\n/g, '\n').replace(/\r/g, '\n');
-
-    // Split on double line breaks, accounting for possible multiple \n characters
-    const sections = normalizedChunk.split(/\n{2,}/);
-
-    // Must have headers and body sections
-    if (sections.length < 2) {
-        return false;
-    }
-
-    const lastSection = sections[sections.length - 1].trim();
-
-    // Skip empty sections
-    if (!lastSection) {
-        return false;
-    }
-
-    // Must have valid JSON body
-    try {
-        JSON.parse(lastSection);
-        return true;
-    } catch {
-        return false;
-    }
-}
+// function parseHeaders(headerLines: string[]): Headers {
+//     const headers: Headers = {};
+//     for (const line of headerLines) {
+//         const [key, value] = line.split(': ');
+//         switch (key.toLowerCase()) {
+//             case 'content-type':
+//                 headers.contentType = value;
+//                 break;
+//             case 'content-length':
+//                 headers.contentLength = Number(value);
+//                 break;
+//         }
+//     }
+//     return headers;
+// }
+
+const CRLF = '\r\n';
 
 export function parseMultipart({
     responseText,
-    state = {lastProcessedLength: 0, buffer: ''},
+    state = {lastProcessedLength: 0},
     boundary = 'boundary',
 }: {
     responseText: string;
     state?: MultipartState;
     boundary?: string;
 }): MultipartResult {
     // Combine buffer with new data
-    const newData = state.buffer + responseText.slice(state.lastProcessedLength);
+    const newData = responseText.slice(state.lastProcessedLength);
 
     if (!newData) {
         return {chunks: [], state};
     }
 
-    // Split on boundary with double dashes
-    const boundaryStr = `--${boundary}`;
-    const parts = newData.split(boundaryStr).filter(Boolean);
+    // Split on boundary with double dashes and CRLF
+    const boundaryStr = `--${boundary}${CRLF}`;
+    const parts = newData.split(boundaryStr);
 
-    let lastCompleteChunkEnd = 0;
-    let currentPosition = 0;
+    let currentPosition = state.lastProcessedLength;
     const chunks: StreamingChunk[] = [];
-    let newBuffer = state.buffer;
 
     for (let i = 0; i < parts.length; i++) {
         const part = parts[i];
         const isLastPart = i === parts.length - 1;
 
-        // Skip the final boundary marker
-        if (part.trim() === '--') {
-            continue;
-        }
+        // Split part into lines by CRLF
+        const lines = part.split(CRLF);
 
-        // Extract JSON content while preserving line endings
-        const contentMatch = part.match(
-            /Content-Type: application\/json\r\n\r\n([\s\S]*?)(?:\r\n)?$/,
-        );
-        if (!contentMatch) {
+        // Find the empty line that separates headers from content
+        const emptyLineIndex = lines.findIndex((line) => line === '');
+        if (emptyLineIndex === -1 || !lines[emptyLineIndex + 1]) {
             if (isLastPart) {
-                newBuffer = part;
                 break;
             }
             continue;
         }
 
-        const jsonContent = contentMatch[1];
-
-        // Calculate accurate position including:
-        // 1. Boundary length with CRLF
-        // 2. Content-Type header length with CRLF
-        // 3. Extra CRLF after headers
-        // 4. Part content length
-        const contentTypeHeader = 'Content-Type: application/json\r\n';
-        const crlfLength = 2; // \r\n
-        currentPosition +=
-            boundaryStr.length +
-            crlfLength + // boundary line
-            contentTypeHeader.length + // Content-Type header
-            crlfLength + // Extra CRLF after headers
-            jsonContent.length + // Actual content
-            crlfLength; // Final CRLF
+        // const headers = parseHeaders(lines.slice(0, emptyLineIndex));
 
-        // If it's the last part and not a complete chunk, store in buffer
-        if (isLastPart && !isCompleteChunk(jsonContent)) {
-            // For partial chunks, we need to store the raw content
-            // without headers since they will be added again in the next iteration
-            newBuffer = jsonContent;
-            break;
-        }
+        const jsonContent = lines[emptyLineIndex + 1];
 
+        // if (headers.contentLength) {
+        //     jsonContent = jsonContent.slice(0, headers.contentLength);
+        // }
+
+        // if (headers.contentLength && jsonContent.length < headers.contentLength) {
+        //     newBuffer = jsonContent;
+        //     break;
+        // }
+
+        let parsedChunk: StreamingChunk | null = null;
         try {
-            const parsedChunk = JSON.parse(jsonContent) as StreamingChunk;
-            chunks.push(parsedChunk);
-            lastCompleteChunkEnd = currentPosition;
-            newBuffer = ''; // Clear buffer after successful chunk parse
-        } catch {
-            if (isLastPart) {
-                newBuffer = jsonContent;
-                break;
-            }
+            parsedChunk = JSON.parse(jsonContent) as StreamingChunk;
+        } catch {}
+
+        // If it's the last part and not a complete chunk, store in buffer
+        if (!parsedChunk) {
+            break;
         }
+        // Track position by adding boundary length and full part length
+        chunks.push(parsedChunk);
+        currentPosition += boundaryStr.length + part.length;
     }
 
     // Update state with precise position tracking
     const newState: MultipartState = {
         // Only update lastProcessedLength if we actually processed some chunks
-        lastProcessedLength:
-            chunks.length > 0
-                ? state.lastProcessedLength + lastCompleteChunkEnd
-                : state.lastProcessedLength,
-        buffer: newBuffer,
+        lastProcessedLength: currentPosition,
     };
 
     return {chunks, state: newState};