fix(ext/node): flush brotli decompression stream (#22856)

Fixes #22259

The decompressed input size was not restored because of improper
flushing of the CBrotliDecompressStream state.

ext/node/ops/zlib/brotli.rs

@@ -314,13 +314,15 @@ pub fn op_brotli_decompress_stream_end(
   unsafe {
     let mut available_out = output.len();
     let mut next_out = output.as_mut_ptr();
+    let mut available_in = 0;
+    let mut next_in = [];
     let mut total_out = 0;
 
     if matches!(
       CBrotliDecoderDecompressStream(
         ctx.inst,
-        &mut 0,
-        std::ptr::null_mut(),
+        &mut available_in,
+        next_in.as_mut_ptr(),
         &mut available_out,
         &mut next_out,
         &mut total_out,

ext/node/polyfills/_brotli.js

@@ -12,6 +12,7 @@ import {
   op_brotli_decompress,
   op_brotli_decompress_async,
   op_brotli_decompress_stream,
+  op_brotli_decompress_stream_end,
   op_create_brotli_compress,
   op_create_brotli_decompress,
 } from "ext:core/ops";
@@ -57,6 +58,11 @@ export class BrotliDecompress extends Transform {
         callback();
       },
       flush(callback) {
+        const output = new Uint8Array(1024);
+        let avail;
+        while ((avail = op_brotli_decompress_stream_end(context, output)) > 0) {
+          this.push(output.slice(0, avail));
+        }
         core.close(context);
         callback();
       },

tests/unit_node/zlib_test.ts

@@ -171,3 +171,13 @@ Deno.test("brotli large chunk size", async () => {
   );
   assertEquals(output.length, input.length);
 });
+
+Deno.test("brotli decompress flush restore size", async () => {
+  const input = new Uint8Array(1000000);
+  const output = await buffer(
+    Readable.from([input])
+      .pipe(createBrotliCompress())
+      .pipe(createBrotliDecompress()),
+  );
+  assertEquals(output.length, input.length);
+});