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

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

6 files changed, 531 insertions, 18 deletions

diff --git a/ext/node/Cargo.toml b/ext/node/Cargo.toml
index 125f58571..75d19e917 100644
--- a/ext/node/Cargo.toml
+++ b/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
diff --git a/ext/node/lib.rs b/ext/node/lib.rs
index d144d89ca..99c138b8f 100644
--- a/ext/node/lib.rs
+++ b/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",
diff --git a/ext/node/ops/zlib/brotli.rs b/ext/node/ops/zlib/brotli.rs
new file mode 100644
index 000000000..f3b5001aa
--- /dev/null
+++ b/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)
+  }
+}
diff --git a/ext/node/ops/zlib/mod.rs b/ext/node/ops/zlib/mod.rs
index c103b3009..3d58d16f9 100644
--- a/ext/node/ops/zlib/mod.rs
+++ b/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;
 
diff --git a/ext/node/polyfills/_brotli.js b/ext/node/polyfills/_brotli.js
new file mode 100644
index 000000000..d200d01b6
--- /dev/null
+++ b/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)));
+}
diff --git a/ext/node/polyfills/zlib.ts b/ext/node/polyfills/zlib.ts
index 07bc65c2d..33f17fc4e 100644
--- a/ext/node/polyfills/zlib.ts
+++ b/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,