fix(ext/node): support brotli APIs (#19223)

Co-authored-by: Bartek Iwańczuk <biwanczuk@gmail.com>

Cargo.lock

@@ -1270,6 +1270,7 @@ name = "deno_node"
 version = "0.46.0"
 dependencies = [
  "aes",
+ "brotli",
  "cbc",
  "data-encoding",
  "deno_core",

Cargo.toml

@@ -83,6 +83,7 @@ async-trait = "0.1.51"
 atty = "=0.2.14"
 base64 = "=0.13.1"
 bencher = "0.1"
+brotli = "3.3.4"
 bytes = "1.4.0"
 cache_control = "=0.2.0"
 cbc = { version = "=0.1.2", features = ["alloc"] }

cli/tests/integration/node_unit_tests.rs

@@ -74,7 +74,8 @@ util::unit_test_factory!(
     tty_test,
     util_test,
     v8_test,
-    worker_threads_test
+    worker_threads_test,
+    zlib_test
   ]
 );
 

cli/tests/unit_node/testdata/lorem_ipsum.txt

@@ -0,0 +1 @@
+Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua. Ut enim ad minim veniam, quis nostrud exercitation ullamco laboris nisi ut aliquip ex ea commodo consequat. Duis aute irure dolor in reprehenderit in voluptate velit esse cillum dolore eu fugiat nulla pariatur. Excepteur sint occaecat cupidatat non proident, sunt in culpa qui officia deserunt mollit anim id est laborum.

cli/tests/unit_node/zlib_test.ts

@@ -0,0 +1,62 @@
+// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
+
+import {
+  assert,
+  assertEquals,
+} from "../../../test_util/std/testing/asserts.ts";
+import { deferred } from "../../../test_util/std/async/deferred.ts";
+import { fromFileUrl, relative } from "../../../test_util/std/path/mod.ts";
+import {
+  brotliCompressSync,
+  brotliDecompressSync,
+  createBrotliCompress,
+  createBrotliDecompress,
+} from "node:zlib";
+import { Buffer } from "node:buffer";
+import { createReadStream, createWriteStream } from "node:fs";
+
+Deno.test("brotli compression sync", () => {
+  const buf = Buffer.from("hello world");
+  const compressed = brotliCompressSync(buf);
+  const decompressed = brotliDecompressSync(compressed);
+  assertEquals(decompressed.toString(), "hello world");
+});
+
+Deno.test("brotli compression", async () => {
+  const promise = deferred();
+  const compress = createBrotliCompress();
+  const filePath = relative(
+    Deno.cwd(),
+    fromFileUrl(new URL("./testdata/lorem_ipsum.txt", import.meta.url)),
+  );
+  const input = createReadStream(filePath);
+  const output = createWriteStream("lorem_ipsum.txt.br");
+
+  const stream = input.pipe(compress).pipe(output);
+
+  stream.on("finish", () => {
+    const decompress = createBrotliDecompress();
+    const input2 = createReadStream("lorem_ipsum.txt.br");
+    const output2 = createWriteStream("lorem_ipsum.txt");
+
+    const stream2 = input2.pipe(decompress).pipe(output2);
+
+    stream2.on("finish", () => {
+      promise.resolve();
+    });
+  });
+
+  await promise;
+  const content = Deno.readTextFileSync("lorem_ipsum.txt");
+  assert(content.startsWith("Lorem ipsum dolor sit amet"));
+  try {
+    Deno.removeSync("lorem_ipsum.txt.br");
+  } catch {
+    // pass
+  }
+  try {
+    Deno.removeSync("lorem_ipsum.txt");
+  } catch {
+    // pass
+  }
+});

ext/node/Cargo.toml

@@ -15,6 +15,7 @@ path = "lib.rs"
 
 [dependencies]
 aes.workspace = true
+brotli.workspace = true
 cbc.workspace = true
 data-encoding = "2.3.3"
 deno_core.workspace = true

ext/node/lib.rs

@@ -211,6 +211,16 @@ deno_core::extension!(deno_node,
     ops::zlib::op_zlib_write_async,
     ops::zlib::op_zlib_init,
     ops::zlib::op_zlib_reset,
+    ops::zlib::brotli::op_brotli_compress,
+    ops::zlib::brotli::op_brotli_compress_async,
+    ops::zlib::brotli::op_create_brotli_compress,
+    ops::zlib::brotli::op_brotli_compress_stream,
+    ops::zlib::brotli::op_brotli_compress_stream_end,
+    ops::zlib::brotli::op_brotli_decompress,
+    ops::zlib::brotli::op_brotli_decompress_async,
+    ops::zlib::brotli::op_create_brotli_decompress,
+    ops::zlib::brotli::op_brotli_decompress_stream,
+    ops::zlib::brotli::op_brotli_decompress_stream_end,
     ops::http::op_node_http_request<P>,
     op_node_build_os,
     ops::require::op_require_init_paths,
@@ -242,6 +252,7 @@ deno_core::extension!(deno_node,
     "00_globals.js",
     "01_require.js",
     "02_init.js",
+    "_brotli.js",
     "_events.mjs",
     "_fs/_fs_access.ts",
     "_fs/_fs_appendFile.ts",

ext/node/ops/zlib/brotli.rs

@@ -0,0 +1,349 @@
+// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
+use brotli::enc::encode::BrotliEncoderParameter;
+use brotli::ffi::compressor::*;
+use brotli::ffi::decompressor::ffi::interface::BrotliDecoderResult;
+use brotli::ffi::decompressor::ffi::BrotliDecoderState;
+use brotli::ffi::decompressor::*;
+use deno_core::error::type_error;
+use deno_core::error::AnyError;
+use deno_core::op;
+use deno_core::JsBuffer;
+use deno_core::OpState;
+use deno_core::Resource;
+use deno_core::ToJsBuffer;
+
+fn encoder_mode(mode: u32) -> Result<BrotliEncoderMode, AnyError> {
+  if mode > 6 {
+    return Err(type_error("Invalid encoder mode"));
+  }
+  // SAFETY: mode is a valid discriminant for BrotliEncoderMode
+  unsafe { Ok(std::mem::transmute::<u32, BrotliEncoderMode>(mode)) }
+}
+
+#[op]
+pub fn op_brotli_compress(
+  buffer: &[u8],
+  out: &mut [u8],
+  quality: i32,
+  lgwin: i32,
+  mode: u32,
+) -> Result<usize, AnyError> {
+  let in_buffer = buffer.as_ptr();
+  let in_size = buffer.len();
+  let out_buffer = out.as_mut_ptr();
+  let mut out_size = out.len();
+
+  // SAFETY: in_size and in_buffer, out_size and out_buffer are valid for this call.
+  if unsafe {
+    BrotliEncoderCompress(
+      quality,
+      lgwin,
+      encoder_mode(mode)?,
+      in_size,
+      in_buffer,
+      &mut out_size as *mut usize,
+      out_buffer,
+    )
+  } != 1
+  {
+    return Err(type_error("Failed to compress"));
+  }
+
+  Ok(out_size)
+}
+
+fn max_compressed_size(input_size: usize) -> usize {
+  if input_size == 0 {
+    return 2;
+  }
+
+  // [window bits / empty metadata] + N * [uncompressed] + [last empty]
+  let num_large_blocks = input_size >> 14;
+  let overhead = 2 + (4 * num_large_blocks) + 3 + 1;
+  let result = input_size + overhead;
+
+  if result < input_size {
+    0
+  } else {
+    result
+  }
+}
+
+#[op]
+pub async fn op_brotli_compress_async(
+  input: JsBuffer,
+  quality: i32,
+  lgwin: i32,
+  mode: u32,
+) -> Result<ToJsBuffer, AnyError> {
+  tokio::task::spawn_blocking(move || {
+    let in_buffer = input.as_ptr();
+    let in_size = input.len();
+
+    let mut out = vec![0u8; max_compressed_size(in_size)];
+    let out_buffer = out.as_mut_ptr();
+    let mut out_size = out.len();
+
+    // SAFETY: in_size and in_buffer, out_size and out_buffer
+    // are valid for this call.
+    if unsafe {
+      BrotliEncoderCompress(
+        quality,
+        lgwin,
+        encoder_mode(mode)?,
+        in_size,
+        in_buffer,
+        &mut out_size as *mut usize,
+        out_buffer,
+      )
+    } != 1
+    {
+      return Err(type_error("Failed to compress"));
+    }
+
+    out.truncate(out_size);
+    Ok(out.into())
+  })
+  .await?
+}
+
+struct BrotliCompressCtx {
+  inst: *mut BrotliEncoderState,
+}
+
+impl Resource for BrotliCompressCtx {}
+
+impl Drop for BrotliCompressCtx {
+  fn drop(&mut self) {
+    // SAFETY: `self.inst` is the current brotli encoder instance.
+    // It is not used after the following call.
+    unsafe { BrotliEncoderDestroyInstance(self.inst) };
+  }
+}
+
+#[op]
+pub fn op_create_brotli_compress(
+  state: &mut OpState,
+  params: Vec<(u8, i32)>,
+) -> u32 {
+  let inst =
+    // SAFETY: Creates a brotli encoder instance for default allocators.
+    unsafe { BrotliEncoderCreateInstance(None, None, std::ptr::null_mut()) };
+
+  for (key, value) in params {
+    // SAFETY: `key` can range from 0-255.
+    // Any valid u32 can be used for the `value`.
+    unsafe {
+      BrotliEncoderSetParameter(inst, encoder_param(key), value as u32);
+    }
+  }
+
+  state.resource_table.add(BrotliCompressCtx { inst })
+}
+
+fn encoder_param(param: u8) -> BrotliEncoderParameter {
+  // SAFETY: BrotliEncoderParam is valid for 0-255
+  unsafe { std::mem::transmute(param as u32) }
+}
+
+#[op]
+pub fn op_brotli_compress_stream(
+  state: &mut OpState,
+  rid: u32,
+  input: &[u8],
+  output: &mut [u8],
+) -> Result<usize, AnyError> {
+  let ctx = state.resource_table.get::<BrotliCompressCtx>(rid)?;
+
+  // SAFETY: TODO(littledivy)
+  unsafe {
+    let mut available_in = input.len();
+    let mut next_in = input.as_ptr();
+    let mut available_out = output.len();
+    let mut next_out = output.as_mut_ptr();
+    let mut total_out = 0;
+
+    if BrotliEncoderCompressStream(
+      ctx.inst,
+      BrotliEncoderOperation::BROTLI_OPERATION_PROCESS,
+      &mut available_in,
+      &mut next_in,
+      &mut available_out,
+      &mut next_out,
+      &mut total_out,
+    ) != 1
+    {
+      return Err(type_error("Failed to compress"));
+    }
+
+    // On progress, next_out is advanced and available_out is reduced.
+    Ok(output.len() - available_out)
+  }
+}
+
+#[op]
+pub fn op_brotli_compress_stream_end(
+  state: &mut OpState,
+  rid: u32,
+  output: &mut [u8],
+) -> Result<usize, AnyError> {
+  let ctx = state.resource_table.take::<BrotliCompressCtx>(rid)?;
+
+  // SAFETY: TODO(littledivy)
+  unsafe {
+    let mut available_out = output.len();
+    let mut next_out = output.as_mut_ptr();
+    let mut total_out = 0;
+
+    if BrotliEncoderCompressStream(
+      ctx.inst,
+      BrotliEncoderOperation::BROTLI_OPERATION_FINISH,
+      &mut 0,
+      std::ptr::null_mut(),
+      &mut available_out,
+      &mut next_out,
+      &mut total_out,
+    ) != 1
+    {
+      return Err(type_error("Failed to compress"));
+    }
+
+    // On finish, next_out is advanced and available_out is reduced.
+    Ok(output.len() - available_out)
+  }
+}
+
+fn brotli_decompress(buffer: &[u8]) -> Result<ToJsBuffer, AnyError> {
+  let in_buffer = buffer.as_ptr();
+  let in_size = buffer.len();
+
+  let mut out = vec![0u8; 4096];
+  loop {
+    let out_buffer = out.as_mut_ptr();
+    let mut out_size = out.len();
+    // SAFETY: TODO(littledivy)
+    match unsafe {
+      CBrotliDecoderDecompress(
+        in_size,
+        in_buffer,
+        &mut out_size as *mut usize,
+        out_buffer,
+      )
+    } {
+      BrotliDecoderResult::BROTLI_DECODER_RESULT_SUCCESS => {
+        out.truncate(out_size);
+        return Ok(out.into());
+      }
+      BrotliDecoderResult::BROTLI_DECODER_RESULT_NEEDS_MORE_OUTPUT => {
+        let new_size = out.len() * 2;
+        if new_size < out.len() {
+          return Err(type_error("Failed to decompress"));
+        }
+        out.resize(new_size, 0);
+      }
+      _ => return Err(type_error("Failed to decompress")),
+    }
+  }
+}
+
+#[op]
+pub fn op_brotli_decompress(buffer: &[u8]) -> Result<ToJsBuffer, AnyError> {
+  brotli_decompress(buffer)
+}
+
+#[op]
+pub async fn op_brotli_decompress_async(
+  buffer: JsBuffer,
+) -> Result<ToJsBuffer, AnyError> {
+  tokio::task::spawn_blocking(move || brotli_decompress(&buffer)).await?
+}
+
+struct BrotliDecompressCtx {
+  inst: *mut BrotliDecoderState,
+}
+
+impl Resource for BrotliDecompressCtx {}
+
+impl Drop for BrotliDecompressCtx {
+  fn drop(&mut self) {
+    // SAFETY: TODO(littledivy)
+    unsafe { CBrotliDecoderDestroyInstance(self.inst) };
+  }
+}
+
+#[op]
+pub fn op_create_brotli_decompress(state: &mut OpState) -> u32 {
+  let inst =
+    // SAFETY: TODO(littledivy)
+    unsafe { CBrotliDecoderCreateInstance(None, None, std::ptr::null_mut()) };
+  state.resource_table.add(BrotliDecompressCtx { inst })
+}
+
+#[op]
+pub fn op_brotli_decompress_stream(
+  state: &mut OpState,
+  rid: u32,
+  input: &[u8],
+  output: &mut [u8],
+) -> Result<usize, AnyError> {
+  let ctx = state.resource_table.get::<BrotliDecompressCtx>(rid)?;
+
+  // SAFETY: TODO(littledivy)
+  unsafe {
+    let mut available_in = input.len();
+    let mut next_in = input.as_ptr();
+    let mut available_out = output.len();
+    let mut next_out = output.as_mut_ptr();
+    let mut total_out = 0;
+
+    if matches!(
+      CBrotliDecoderDecompressStream(
+        ctx.inst,
+        &mut available_in,
+        &mut next_in,
+        &mut available_out,
+        &mut next_out,
+        &mut total_out,
+      ),
+      BrotliDecoderResult::BROTLI_DECODER_RESULT_ERROR
+    ) {
+      return Err(type_error("Failed to decompress"));
+    }
+
+    // On progress, next_out is advanced and available_out is reduced.
+    Ok(output.len() - available_out)
+  }
+}
+
+#[op]
+pub fn op_brotli_decompress_stream_end(
+  state: &mut OpState,
+  rid: u32,
+  output: &mut [u8],
+) -> Result<usize, AnyError> {
+  let ctx = state.resource_table.get::<BrotliDecompressCtx>(rid)?;
+
+  // SAFETY: TODO(littledivy)
+  unsafe {
+    let mut available_out = output.len();
+    let mut next_out = output.as_mut_ptr();
+    let mut total_out = 0;
+
+    if matches!(
+      CBrotliDecoderDecompressStream(
+        ctx.inst,
+        &mut 0,
+        std::ptr::null_mut(),
+        &mut available_out,
+        &mut next_out,
+        &mut total_out,
+      ),
+      BrotliDecoderResult::BROTLI_DECODER_RESULT_ERROR
+    ) {
+      return Err(type_error("Failed to decompress"));
+    }
+
+    // On finish, next_out is advanced and available_out is reduced.
+    Ok(output.len() - available_out)
+  }
+}

ext/node/ops/zlib/mod.rs

@@ -12,6 +12,7 @@ use std::future::Future;
 use std::rc::Rc;
 
 mod alloc;
+pub mod brotli;
 mod mode;
 mod stream;
 

ext/node/polyfills/_brotli.js

@@ -0,0 +1,145 @@
+// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
+
+import { zlib as constants } from "ext:deno_node/internal_binding/constants.ts";
+import { TextEncoder } from "ext:deno_web/08_text_encoding.js";
+import { Transform } from "ext:deno_node/stream.ts";
+import { Buffer } from "ext:deno_node/buffer.ts";
+
+const { core } = globalThis.__bootstrap;
+const { ops } = core;
+
+const enc = new TextEncoder();
+const toU8 = (input) => {
+  if (typeof input === "string") {
+    return enc.encode(input);
+  }
+
+  return input;
+};
+
+export function createBrotliCompress(options) {
+  return new BrotliCompress(options);
+}
+
+export function createBrotliDecompress(options) {
+  return new BrotliDecompress(options);
+}
+
+export class BrotliDecompress extends Transform {
+  #context;
+
+  // TODO(littledivy): use `options` argument
+  constructor(_options = {}) {
+    super({
+      // TODO(littledivy): use `encoding` argument
+      transform(chunk, _encoding, callback) {
+        const input = toU8(chunk);
+        const output = new Uint8Array(1024);
+        const avail = ops.op_brotli_decompress_stream(context, input, output);
+        this.push(output.slice(0, avail));
+        callback();
+      },
+      flush(callback) {
+        core.close(context);
+        callback();
+      },
+    });
+
+    this.#context = ops.op_create_brotli_decompress();
+    const context = this.#context;
+  }
+}
+
+export class BrotliCompress extends Transform {
+  #context;
+
+  constructor(options = {}) {
+    super({
+      // TODO(littledivy): use `encoding` argument
+      transform(chunk, _encoding, callback) {
+        const input = toU8(chunk);
+        const output = new Uint8Array(brotliMaxCompressedSize(input.length));
+        const avail = ops.op_brotli_compress_stream(context, input, output);
+        this.push(output.slice(0, avail));
+        callback();
+      },
+      flush(callback) {
+        const output = new Uint8Array(1024);
+        const avail = ops.op_brotli_compress_stream_end(context, output);
+        this.push(output.slice(0, avail));
+        callback();
+      },
+    });
+
+    const params = Object.values(options?.params ?? {});
+    this.#context = ops.op_create_brotli_compress(params);
+    const context = this.#context;
+  }
+}
+
+function oneOffCompressOptions(options) {
+  const quality = options?.params?.[constants.BROTLI_PARAM_QUALITY] ??
+    constants.BROTLI_DEFAULT_QUALITY;
+  const lgwin = options?.params?.[constants.BROTLI_PARAM_LGWIN] ??
+    constants.BROTLI_DEFAULT_WINDOW;
+  const mode = options?.params?.[constants.BROTLI_PARAM_MODE] ??
+    constants.BROTLI_MODE_GENERIC;
+
+  return {
+    quality,
+    lgwin,
+    mode,
+  };
+}
+
+function brotliMaxCompressedSize(input) {
+  if (input == 0) return 2;
+
+  // [window bits / empty metadata] + N * [uncompressed] + [last empty]
+  const numLargeBlocks = input >> 24;
+  const overhead = 2 + (4 * numLargeBlocks) + 3 + 1;
+  const result = input + overhead;
+
+  return result < input ? 0 : result;
+}
+
+export function brotliCompress(
+  input,
+  options,
+  callback,
+) {
+  const buf = toU8(input);
+
+  if (typeof options === "function") {
+    callback = options;
+    options = {};
+  }
+
+  const { quality, lgwin, mode } = oneOffCompressOptions(options);
+  core.opAsync("op_brotli_compress_async", buf, quality, lgwin, mode)
+    .then((result) => callback(null, result))
+    .catch((err) => callback(err));
+}
+
+export function brotliCompressSync(
+  input,
+  options,
+) {
+  const buf = toU8(input);
+  const output = new Uint8Array(brotliMaxCompressedSize(buf.length));
+
+  const { quality, lgwin, mode } = oneOffCompressOptions(options);
+  const len = ops.op_brotli_compress(buf, output, quality, lgwin, mode);
+  return Buffer.from(output.subarray(0, len));
+}
+
+export function brotliDecompress(input) {
+  const buf = toU8(input);
+  return ops.op_brotli_decompress_async(buf)
+    .then((result) => callback(null, Buffer.from(result)))
+    .catch((err) => callback(err));
+}
+
+export function brotliDecompressSync(input) {
+  return Buffer.from(ops.op_brotli_decompress(toU8(input)));
+}

ext/node/polyfills/zlib.ts

@@ -32,11 +32,29 @@ import {
   unzip,
   unzipSync,
 } from "ext:deno_node/_zlib.mjs";
+import {
+  brotliCompress,
+  brotliCompressSync,
+  brotliDecompress,
+  brotliDecompressSync,
+  createBrotliCompress,
+  createBrotliDecompress,
+} from "ext:deno_node/_brotli.js";
+
 export class Options {
   constructor() {
     notImplemented("Options.prototype.constructor");
   }
 }
+
+interface IBrotliOptions {
+  flush?: number;
+  finishFlush?: number;
+  chunkSize?: number;
+  params?: Record<number, number>;
+  maxOutputLength?: number;
+}
+
 export class BrotliOptions {
   constructor() {
     notImplemented("BrotliOptions.prototype.constructor");
@@ -58,24 +76,6 @@ export class ZlibBase {
   }
 }
 export { constants };
-export function createBrotliCompress() {
-  notImplemented("createBrotliCompress");
-}
-export function createBrotliDecompress() {
-  notImplemented("createBrotliDecompress");
-}
-export function brotliCompress() {
-  notImplemented("brotliCompress");
-}
-export function brotliCompressSync() {
-  notImplemented("brotliCompressSync");
-}
-export function brotliDecompress() {
-  notImplemented("brotliDecompress");
-}
-export function brotliDecompressSync() {
-  notImplemented("brotliDecompressSync");
-}
 
 export default {
   Options,
@@ -122,7 +122,13 @@ export default {
 };
 
 export {
+  brotliCompress,
+  brotliCompressSync,
+  brotliDecompress,
+  brotliDecompressSync,
   codes,
+  createBrotliCompress,
+  createBrotliDecompress,
   createDeflate,
   createDeflateRaw,
   createGunzip,

tools/lint.js

@@ -100,6 +100,7 @@ async function dlintPreferPrimordials() {
     // "ext/node/polyfills/*.mjs",
     // "ext/node/polyfills/*.ts",
     // ":!:ext/node/polyfills/*.d.ts",
+    ":!:ext/node/polyfills/_brotli.js",
     "core/*.js",
     ":!:core/*_test.js",
     ":!:core/examples/**",