fix(core): stop dropping the first chunk of ffmpeg output in read_lines

packages/core/src/lib/file_processor.ts

@@ -513,7 +513,7 @@ class FileProcessor {
 
         let error_string = ''
         // TODO use the input timestamp in a progress meter
-        for await (const line of this.#readlines(proc.stderr)) {
+        for await (const line of FileProcessor.read_lines(proc.stderr)) {
           if (line.includes('Parsed_showinfo') && line.includes(' n: ')) {
             const pts_time_str = line.match(/pts_time:(?<pts_time>\d+([.]\d*)?)/)?.groups?.pts_time
             if (pts_time_str === undefined) {
@@ -649,22 +649,22 @@ class FileProcessor {
     }
   }
 
-  async * #readlines(stream: ReadableStream<Uint8Array>) {
+  static async * read_lines(stream: ReadableStream<Uint8Array>) {
     const reader = stream.pipeThrough(new TextDecoderStream()).getReader()
     let buffer = ''
-    let read_result = await reader.read()
-    while (!read_result.done) {
-      read_result = await reader.read()
+    while (true) {
+      const read_result = await reader.read()
+      if (read_result.done) {
+        break
+      }
       buffer += read_result.value ?? ''
       const lines = buffer.split('\n')
 
       // ignore the last line because we dont know where its newline is yet
       for (let i = 0; i < lines.length - 1; i++) {
         yield lines[i]
       }
-      if (lines.length > 0) {
-        buffer = lines.at(-1)!
-      }
+      buffer = lines.at(-1)!
     }
 
     if (buffer.length > 0) {

packages/core/test/file_processor.test.ts

@@ -0,0 +1,48 @@
+import { test } from 'forager-test'
+import { FileProcessor } from '~/lib/file_processor.ts'
+
+function stream_of(chunks: string[]): ReadableStream<Uint8Array> {
+  const encoder = new TextEncoder()
+  return new ReadableStream({
+    start(controller) {
+      for (const chunk of chunks) {
+        controller.enqueue(encoder.encode(chunk))
+      }
+      controller.close()
+    }
+  })
+}
+
+test('FileProcessor.read_lines', async (ctx) => {
+  await ctx.subtest('yields every line, including lines in the first chunk', async () => {
+    const lines = await Array.fromAsync(FileProcessor.read_lines(stream_of([
+      'first\nsecond\n',
+      'third\nfourth\n',
+    ])))
+    ctx.assert.equals(lines, ['first', 'second', 'third', 'fourth'])
+  })
+
+  await ctx.subtest('reads a single chunk stream', async () => {
+    const lines = await Array.fromAsync(FileProcessor.read_lines(stream_of(['only line\n'])))
+    ctx.assert.equals(lines, ['only line'])
+  })
+
+  await ctx.subtest('joins lines split across chunk boundaries', async () => {
+    const lines = await Array.fromAsync(FileProcessor.read_lines(stream_of([
+      'frame ',
+      'one\nframe ',
+      'two',
+    ])))
+    ctx.assert.equals(lines, ['frame one', 'frame two'])
+  })
+
+  await ctx.subtest('yields a trailing line without a newline', async () => {
+    const lines = await Array.fromAsync(FileProcessor.read_lines(stream_of(['first\nno trailing newline'])))
+    ctx.assert.equals(lines, ['first', 'no trailing newline'])
+  })
+
+  await ctx.subtest('yields nothing for an empty stream', async () => {
+    const lines = await Array.fromAsync(FileProcessor.read_lines(stream_of([])))
+    ctx.assert.equals(lines, [])
+  })
+})