runtime: implement CompressionStream/DecompressionStream (#24757)

Closes https://github.com/oven-sh/bun/issues/1723
Closes https://github.com/oven-sh/bun/pull/22214
Closes https://github.com/oven-sh/bun/pull/24241

also supports the `"brotli"` and `"zstd"` formats

<img width="1244" height="547" alt="image"
src="https://github.com/user-attachments/assets/aecf4489-29ad-411d-9f6b-3bee50ed1b27"
/>

test/integration/bun-types/fixture/streams.ts

@@ -59,6 +59,8 @@ expectType(node_stream.blob()).is<Promise<Blob>>();
 
 Bun.file("./foo.csv").stream().pipeThrough(new TextDecoderStream()).pipeThrough(new TextEncoderStream());
 
+Bun.file("./foo.csv").stream().pipeThrough(new CompressionStream("gzip")).pipeThrough(new DecompressionStream("gzip"));
+
 Bun.file("./foo.csv")
   .stream()
   .pipeThrough(new TextDecoderStream())

test/js/node/test/parallel/test-global-webstreams.js

@@ -0,0 +1,24 @@
+'use strict';
+
+require('../common');
+
+const assert = require('assert');
+const webstreams = require('stream/web');
+
+assert.strictEqual(ReadableStream, webstreams.ReadableStream);
+assert.strictEqual(ReadableStreamDefaultReader, webstreams.ReadableStreamDefaultReader);
+assert.strictEqual(ReadableStreamBYOBReader, webstreams.ReadableStreamBYOBReader);
+assert.strictEqual(ReadableStreamBYOBRequest, webstreams.ReadableStreamBYOBRequest);
+assert.strictEqual(ReadableByteStreamController, webstreams.ReadableByteStreamController);
+assert.strictEqual(ReadableStreamDefaultController, webstreams.ReadableStreamDefaultController);
+assert.strictEqual(TransformStream, webstreams.TransformStream);
+assert.strictEqual(TransformStreamDefaultController, webstreams.TransformStreamDefaultController);
+assert.strictEqual(WritableStream, webstreams.WritableStream);
+assert.strictEqual(WritableStreamDefaultWriter, webstreams.WritableStreamDefaultWriter);
+assert.strictEqual(WritableStreamDefaultController, webstreams.WritableStreamDefaultController);
+assert.strictEqual(ByteLengthQueuingStrategy, webstreams.ByteLengthQueuingStrategy);
+assert.strictEqual(CountQueuingStrategy, webstreams.CountQueuingStrategy);
+assert.strictEqual(TextEncoderStream, webstreams.TextEncoderStream);
+assert.strictEqual(TextDecoderStream, webstreams.TextDecoderStream);
+assert.strictEqual(CompressionStream, webstreams.CompressionStream);
+assert.strictEqual(DecompressionStream, webstreams.DecompressionStream);

test/js/node/test/parallel/test-whatwg-webstreams-compression.js

@@ -0,0 +1,70 @@
+// Flags: --no-warnings
+'use strict';
+
+const common = require('../common');
+
+const {
+  CompressionStream,
+  DecompressionStream,
+} = require('stream/web');
+
+const assert = require('assert');
+const dec = new TextDecoder();
+
+async function test(format) {
+  const gzip = new CompressionStream(format);
+  const gunzip = new DecompressionStream(format);
+
+  assert.strictEqual(gzip[Symbol.toStringTag], 'CompressionStream');
+  assert.strictEqual(gunzip[Symbol.toStringTag], 'DecompressionStream');
+
+  gzip.readable.pipeTo(gunzip.writable).then(common.mustCall());
+
+  const reader = gunzip.readable.getReader();
+  const writer = gzip.writable.getWriter();
+
+  const compressed_data = [];
+  const reader_function = ({ value, done }) => {
+    if (value)
+      compressed_data.push(value);
+    if (!done)
+      return reader.read().then(reader_function);
+    assert.strictEqual(dec.decode(Buffer.concat(compressed_data)), 'hello');
+  };
+  const reader_promise = reader.read().then(reader_function);
+
+  await Promise.all([
+    reader_promise,
+    reader_promise.then(() => reader.read().then(({ done }) => assert(done))),
+    writer.write('hello'),
+    writer.close(),
+  ]);
+}
+
+Promise.all(['gzip', 'deflate', 'deflate-raw', 'brotli', 'zstd'].map((i) => test(i))).then(common.mustCall());
+
+[1, 'hello', false, {}].forEach((i) => {
+  assert.throws(() => new CompressionStream(i), {
+    code: 'ERR_INVALID_ARG_VALUE',
+  });
+  assert.throws(() => new DecompressionStream(i), {
+    code: 'ERR_INVALID_ARG_VALUE',
+  });
+});
+
+assert.throws(
+  () => Reflect.get(CompressionStream.prototype, 'readable', {}), {
+    name: 'TypeError',
+  });
+assert.throws(
+  () => Reflect.get(CompressionStream.prototype, 'writable', {}), {
+    name: 'TypeError',
+  });
+assert.throws(
+  () => Reflect.get(DecompressionStream.prototype, 'readable', {}), {
+    name: 'TypeError',
+  });
+assert.throws(
+  () => Reflect.get(DecompressionStream.prototype, 'writable', {}), {
+    name: 'TypeError',
+  });