/**
 *
 * Agora Real Time Engagement
 * Created by Wei Hu in 2021-09.
 * Copyright (c) 2022 Agora IO. All rights reserved.
 *
 */
#include "rte_runtime/binding/nodejs/src/extension/extension.h"

#include <assert.h>
#include <stdbool.h>
#include <stdlib.h>
#include <string.h>

#include "lib/alloc.h"
#include "lib/list.h"
#include "lib/list_node.h"
#include "lib/mutex.h"
#include "lib/shared_ptr.h"
#include "lib/signature.h"
#include "lib/string.h"
#include "rte_runtime/binding/nodejs/src/common/common.h"
#include "rte_runtime/binding/nodejs/src/common/semaphore.h"
#include "rte_runtime/binding/nodejs/src/common/tsfn.h"
#include "rte_runtime/binding/nodejs/src/metadata/metadata.h"
#include "rte_runtime/binding/nodejs/src/msg/cmd/cmd.h"
#include "rte_runtime/binding/nodejs/src/msg/image_frame/image_frame.h"
#include "rte_runtime/binding/nodejs/src/msg/pcm_frame/pcm_frame.h"
#include "rte_runtime/binding/nodejs/src/rte/rte.h"
#include "rte_runtime/extension/extension.h"
#include "rte_runtime/extension_thread/extension_thread.h"
#include "rte_runtime/metadata/metadata.h"
#include "rte_runtime/msg/cmd/cmd.h"
#include "rte_runtime/msg/data/data.h"
#include "rte_runtime/rte/rte.h"
#include "rte_runtime/sanitizer/thread_check.h"

#define ZF_LOG_TAG "Node"
#include "zf_log.h"

#define GET_JS_EXTENSION_AND_JS_RTE_FROM_BRIDGE(env, bridge, js_extension,     \
                                                js_rte)                        \
  do {                                                                         \
    napi_status status =                                                       \
        napi_get_reference_value((env), (bridge)->js_ref, &(js_extension));    \
    GOTO_LABEL_IF_NAPI_FAIL(done, status == napi_ok && js_extension != NULL,   \
                            "Failed to get JS extension: %d", status);         \
                                                                               \
    status = napi_get_reference_value((env), (bridge)->js_rte_ref, &(js_rte)); \
    GOTO_LABEL_IF_NAPI_FAIL(done, status == napi_ok && js_rte != NULL,         \
                            "Failed to get JS rte: %d", status);               \
  } while (0)

typedef struct on_init_call_info_t {
  rte_nodejs_extension_t* extension_bridge;
  rte_t* rte;
  rte_metadata_info_t* manifest;
  rte_metadata_info_t* property;
  rte_nodejs_sem_t sem;
  bool js_result;
} on_init_call_info_t;

typedef struct on_start_stop_call_info_t {
  rte_nodejs_extension_t* extension_bridge;
  rte_nodejs_sem_t sem;
  bool js_result;
} on_start_stop_call_info_t;

// shared by on_cmd, on_data, on_pcm_frame, on_image_frame
typedef struct on_xxx_call_info_t {
  rte_nodejs_extension_t* extension_bridge;
  rte_sharedptr_t* msg;
  rte_nodejs_sem_t sem;
  bool js_result;
} on_xxx_call_info_t;

bool rte_nodejs_extension_check_integrity(rte_nodejs_extension_t* self,
                                          bool check_thread) {
  assert(self);

  if (rte_signature_get(&self->signature) != RTE_NODEJS_EXTENSION_SIGNATURE) {
    return false;
  }
  if (check_thread &&
      !rte_sanitizer_thread_check_do_check(&self->thread_check)) {
    return false;
  }
  return true;
}

static void invoke_extension_js_on_init(napi_env env, napi_value fn,
                                        UNUSED void* context, void* data) {
  assert(env && fn);

  on_init_call_info_t* call_info = data;
  assert(call_info);

  rte_nodejs_rte_t* rte_bridge = NULL;
  napi_value js_rte = rte_nodejs_rte_wrap(env, call_info->rte, &rte_bridge);
  assert(js_rte);
  assert(rte_bridge && rte_nodejs_rte_check_integrity(rte_bridge, true));

  assert(call_info->manifest &&
         rte_metadata_info_check_integrity(call_info->manifest));
  napi_value js_manifest =
      rte_nodejs_metadata_info_wrap(env, call_info->manifest);
  assert(js_manifest);

  assert(call_info->property &&
         rte_metadata_info_check_integrity(call_info->property));
  napi_value js_property =
      rte_nodejs_metadata_info_wrap(env, call_info->property);
  assert(js_property);

  napi_status status = napi_ok;

  // There might be no JS references to the JS 'rte' object in the JS world,
  // so we need to take a reference of it.
  if (!call_info->extension_bridge->js_rte_ref) {
    status = napi_create_reference(env, js_rte, 1,
                                   &call_info->extension_bridge->js_rte_ref);
    GOTO_LABEL_IF_NAPI_FAIL(error, status == napi_ok,
                            "Failed to create C reference to JS 'rte': %d",
                            status);
  }

  napi_value js_extension = NULL;
  status = napi_get_reference_value(env, call_info->extension_bridge->js_ref,
                                    &js_extension);
  GOTO_LABEL_IF_NAPI_FAIL(error, status == napi_ok && js_extension != NULL,
                          "Failed to get JS extension: %d", status);

  napi_value result = NULL;
  napi_value argv[] = {js_rte, js_manifest, js_property};
  status = napi_call_function(env, js_extension, fn, 3, argv, &result);
  GOTO_LABEL_IF_NAPI_FAIL(error, status == napi_ok,
                          "Failed to call JS extension on_init(): %d", status);

  goto done;

error:
  call_info->js_result = false;

  rte_nodejs_metadata_info_invalidate_js(env, js_manifest);
  rte_nodejs_metadata_info_invalidate_js(env, js_property);

done:
  rte_nodejs_sem_signal(&call_info->sem);
}

static void proxy_on_init(rte_extension_t* extension, rte_t* rte,
                          rte_metadata_info_t* manifest,
                          rte_metadata_info_t* property) {
  assert(extension && rte);

  rte_nodejs_extension_t* extension_bridge =
      rte_extension_get_me_in_target_lang(extension);
  assert(extension_bridge &&
         // RTE_NOLINTNEXTLINE(thread-check)
         // thread-check: in order to maintain thread safety, we use semaphore
         // below to prevent JS main thread and the RTE extension thread access
         // the extension bridge at the same time.
         rte_nodejs_extension_check_integrity(extension_bridge, false));

  on_init_call_info_t call_info = {
      extension_bridge, rte, manifest, property, {0}, .js_result = true,
  };

  PREPARE_TO_CALL_TSFN(call_info);

  bool rc = rte_nodejs_threadsafe_function_invoke(extension_bridge->js_on_init,
                                                  &call_info);

  END_TO_CALL_TSFN(call_info, "Failed to call extension on_init().");

  if (!rc || !call_info.js_result) {
    rte_on_init_done(rte, manifest, property);
  }
}

static void invoke_extension_js_on_start(napi_env env, napi_value fn,
                                         UNUSED void* context, void* data) {
  on_start_stop_call_info_t* call_info = data;
  assert(call_info);

  rte_nodejs_extension_t* extension_bridge = call_info->extension_bridge;
  assert(extension_bridge &&
         rte_nodejs_extension_check_integrity(extension_bridge, true));

  napi_value js_extension = NULL;
  napi_value js_rte = NULL;
  GET_JS_EXTENSION_AND_JS_RTE_FROM_BRIDGE(env, extension_bridge, js_extension,
                                          js_rte);

  napi_value result = NULL;
  napi_status status =
      napi_call_function(env, js_extension, fn, 1, &js_rte, &result);
  GOTO_LABEL_IF_NAPI_FAIL(error, status == napi_ok,
                          "Failed to call JS extension on_start().");

  goto done;

error:
  call_info->js_result = false;

done:
  rte_nodejs_sem_signal(&call_info->sem);
}

static void rte_nodejs_extension_release_callbacks(
    napi_env env, rte_nodejs_extension_t* extension_bridge) {
  assert(env && extension_bridge &&
         rte_nodejs_extension_check_integrity(extension_bridge, true));

  rte_nodejs_threadsafe_function_release(env, extension_bridge->js_on_init);
  rte_nodejs_threadsafe_function_release(env, extension_bridge->js_on_start);
  rte_nodejs_threadsafe_function_release(env, extension_bridge->js_on_stop);
  rte_nodejs_threadsafe_function_release(env, extension_bridge->js_on_cmd);
  rte_nodejs_threadsafe_function_release(env, extension_bridge->js_on_data);
  rte_nodejs_threadsafe_function_release(env,
                                         extension_bridge->js_on_pcm_frame);
  rte_nodejs_threadsafe_function_release(env,
                                         extension_bridge->js_on_image_frame);
}

static void rte_nodejs_extension_release_send_xxx_callbacks(
    napi_env env, rte_nodejs_extension_t* extension_bridge) {
  assert(env && extension_bridge &&
         rte_nodejs_extension_check_integrity(extension_bridge, true));
  assert(rte_extension_thread_not_call_by_me(
      extension_bridge->c_extension->extension_thread));

  rte_nodejs_rte_t* rte_bridge =
      extension_bridge->c_extension->rte->me_in_target_lang;
  assert(rte_bridge && rte_nodejs_rte_check_integrity(rte_bridge, true));

  rte_list_foreach (&rte_bridge->tsfns, iter) {
    rte_nodejs_threadsafe_function_t* rte_tsfn =
        rte_ptr_listnode_get(iter.node);
    assert(rte_tsfn);

    rte_nodejs_threadsafe_function_release(env, rte_tsfn);

    rte_listnode_destroy(iter.node);
  }
}

static void invoke_extension_js_on_stop(napi_env env, napi_value fn,
                                        UNUSED void* context, void* data) {
  on_start_stop_call_info_t* call_info = data;
  assert(call_info);

  rte_nodejs_extension_t* extension_bridge = call_info->extension_bridge;
  assert(extension_bridge &&
         rte_nodejs_extension_check_integrity(extension_bridge, true));

  assert(rte_extension_thread_not_call_by_me(
      extension_bridge->c_extension->extension_thread));

  // Release those unfinished JS callbacks for sendXxx to prevent memory leaks
  // and prevent nodejs from exiting.
  rte_nodejs_extension_release_send_xxx_callbacks(env, extension_bridge);

  napi_value js_extension = NULL;
  napi_value js_rte = NULL;
  GET_JS_EXTENSION_AND_JS_RTE_FROM_BRIDGE(env, extension_bridge, js_extension,
                                          js_rte);

  napi_value result = NULL;
  napi_status status =
      napi_call_function(env, js_extension, fn, 1, &js_rte, &result);
  GOTO_LABEL_IF_NAPI_FAIL(error, status == napi_ok,
                          "Failed to call JS extension on_stop().");

  goto done;

error:
  call_info->js_result = false;

done:
  rte_nodejs_sem_signal(&call_info->sem);
}

static void proxy_on_start(rte_extension_t* extension, UNUSED rte_t* rte) {
  rte_nodejs_extension_t* extension_bridge =
      rte_extension_get_me_in_target_lang(extension);
  assert(extension_bridge &&
         // RTE_NOLINTNEXTLINE(thread-check)
         // thread-check: in order to maintain thread safety, we use semaphore
         // below to prevent JS main thread and the RTE extension thread access
         // the extension bridge at the same time.
         rte_nodejs_extension_check_integrity(extension_bridge, false));

  on_start_stop_call_info_t call_info = {
      extension_bridge,
      {0},
      .js_result = true,
  };

  PREPARE_TO_CALL_TSFN(call_info);

  bool rc = rte_nodejs_threadsafe_function_invoke(extension_bridge->js_on_start,
                                                  &call_info);

  END_TO_CALL_TSFN(call_info, "Failed to call extension on_start().");

  if (!rc || !call_info.js_result) {
    rte_on_start_done(rte);
  }
}

static void proxy_on_stop(rte_extension_t* extension, UNUSED rte_t* rte) {
  rte_nodejs_extension_t* extension_bridge =
      rte_extension_get_me_in_target_lang(extension);
  assert(extension_bridge &&
         // RTE_NOLINTNEXTLINE(thread-check)
         // thread-check: in order to maintain thread safety, we use semaphore
         // below to prevent JS main thread and the RTE extension thread access
         // the extension bridge at the same time.
         rte_nodejs_extension_check_integrity(extension_bridge, false));

  on_start_stop_call_info_t call_info = {
      extension_bridge,
      {0},
      .js_result = true,
  };

  PREPARE_TO_CALL_TSFN(call_info);

  bool rc = rte_nodejs_threadsafe_function_invoke(extension_bridge->js_on_stop,
                                                  &call_info);

  END_TO_CALL_TSFN(call_info, "Failed to call extension on_stop().");

  if (!rc || !call_info.js_result) {
    rte_on_stop_done(rte);
  }
}

static void invoke_extension_js_on_cmd(napi_env env, napi_value fn,
                                       UNUSED void* context, void* data) {
  on_xxx_call_info_t* call_info = data;
  assert(call_info);

  napi_value js_extension = NULL;
  napi_value js_rte = NULL;
  GET_JS_EXTENSION_AND_JS_RTE_FROM_BRIDGE(env, call_info->extension_bridge,
                                          js_extension, js_rte);

  napi_value js_cmd = rte_nodejs_cmd_wrap(env, call_info->msg);
  assert(js_cmd);

  napi_value result = NULL;
  napi_value argv[] = {js_rte, js_cmd};
  napi_status status =
      napi_call_function(env, js_extension, fn, 2, argv, &result);
  GOTO_LABEL_IF_NAPI_FAIL(error, status == napi_ok,
                          "Failed to call JS extension on_start().");

  goto done;

error:
  call_info->js_result = false;

done:
  rte_nodejs_sem_signal(&call_info->sem);
}

static void proxy_on_cmd(rte_extension_t* extension, UNUSED rte_t* rte,
                         rte_sharedptr_t* cmd) {
  rte_nodejs_extension_t* extension_bridge =
      rte_extension_get_me_in_target_lang(extension);
  assert(extension_bridge &&
         // RTE_NOLINTNEXTLINE(thread-check)
         // thread-check: in order to maintain thread safety, we use semaphore
         // below to prevent JS main thread and the RTE extension thread access
         // the extension bridge at the same time.
         rte_nodejs_extension_check_integrity(extension_bridge, false));

  on_xxx_call_info_t call_info = {
      extension_bridge,
      cmd,
      {0},
      .js_result = true,
  };

  PREPARE_TO_CALL_TSFN(call_info);

  bool rc = rte_nodejs_threadsafe_function_invoke(extension_bridge->js_on_cmd,
                                                  &call_info);

  END_TO_CALL_TSFN(call_info, "Failed to call extension on_cmd().");

  if (!rc || !call_info.js_result) {
    rte_string_t cmd_name;
    rte_string_init_with_value(&cmd_name, "%s", rte_msg_get_cmd_name(cmd));
    rte_on_cmd_done(rte, rte_cmd_get_cmd_id(cmd), &cmd_name, NULL);
    rte_string_deinit(&cmd_name);
  }
}

static void invoke_extension_js_on_data(napi_env env, napi_value fn,
                                        UNUSED void* context, void* data) {
  on_xxx_call_info_t* call_info = data;

  napi_value js_extension = NULL;
  napi_value js_rte = NULL;
  GET_JS_EXTENSION_AND_JS_RTE_FROM_BRIDGE(env, call_info->extension_bridge,
                                          js_extension, js_rte);

  napi_value js_data = NULL;
  size_t buf_size = rte_data_get_buf_size(call_info->msg);
  void* buf = NULL;  // the data pointer in js ArrayBuffer
  napi_status status = napi_create_arraybuffer(env, buf_size, &buf, &js_data);
  GOTO_LABEL_IF_NAPI_FAIL(done, status == napi_ok,
                          "Failed to create JS arraybuffer: %d", status);

  memcpy(buf, rte_data_get_buf(call_info->msg), buf_size);

  napi_value result = NULL;
  napi_value argv[] = {js_rte, js_data};
  status = napi_call_function(env, js_extension, fn, 2, argv, &result);
  GOTO_LABEL_IF_NAPI_FAIL(error, status == napi_ok,
                          "Failed to call JS extension on_data(): %d", status);

  goto done;

error:
  call_info->js_result = false;

done:
  rte_nodejs_sem_signal(&call_info->sem);
}

static void proxy_on_data(rte_extension_t* extension, UNUSED rte_t* rte,
                          rte_sharedptr_t* data) {
  rte_nodejs_extension_t* extension_bridge =
      rte_extension_get_me_in_target_lang(extension);
  assert(extension_bridge &&
         // RTE_NOLINTNEXTLINE(thread-check)
         // thread-check: in order to maintain thread safety, we use semaphore
         // below to prevent JS main thread and the RTE extension thread access
         // the extension bridge at the same time.
         rte_nodejs_extension_check_integrity(extension_bridge, false));

  on_xxx_call_info_t call_info = {
      extension_bridge,
      data,
      {0},
      .js_result = true,
  };

  PREPARE_TO_CALL_TSFN(call_info);

  bool rc = rte_nodejs_threadsafe_function_invoke(extension_bridge->js_on_data,
                                                  &call_info);

  END_TO_CALL_TSFN(call_info, "Failed to call extension on_data().");

  if (!rc || !call_info.js_result) {
    rte_on_data_done(rte, NULL);
  }
}

static void invoke_extension_js_on_pcm_frame(napi_env env, napi_value fn,
                                             UNUSED void* context, void* data) {
  on_xxx_call_info_t* call_info = data;

  napi_value js_extension = NULL;
  napi_value js_rte = NULL;
  GET_JS_EXTENSION_AND_JS_RTE_FROM_BRIDGE(env, call_info->extension_bridge,
                                          js_extension, js_rte);

  napi_value js_pcm_frame = rte_nodejs_pcm_frame_wrap(env, call_info->msg);
  assert(js_pcm_frame);

  napi_value result = NULL;
  napi_value argv[] = {js_rte, js_pcm_frame};
  napi_status status =
      napi_call_function(env, js_extension, fn, 2, argv, &result);
  GOTO_LABEL_IF_NAPI_FAIL(error, status == napi_ok,
                          "Failed to call JS extension on_pcm_frame().");

  goto done;

error:
  call_info->js_result = false;

done:
  rte_nodejs_sem_signal(&call_info->sem);
}

static void proxy_on_pcm_frame(rte_extension_t* extension, UNUSED rte_t* rte,
                               rte_sharedptr_t* frame) {
  rte_nodejs_extension_t* extension_bridge =
      rte_extension_get_me_in_target_lang(extension);
  assert(extension_bridge &&
         // RTE_NOLINTNEXTLINE(thread-check)
         // thread-check: in order to maintain thread safety, we use semaphore
         // below to prevent JS main thread and the RTE extension thread access
         // the extension bridge at the same time.
         rte_nodejs_extension_check_integrity(extension_bridge, false));

  on_xxx_call_info_t call_info = {
      extension_bridge,
      frame,
      {0},
      .js_result = true,
  };

  PREPARE_TO_CALL_TSFN(call_info);

  bool rc = rte_nodejs_threadsafe_function_invoke(
      extension_bridge->js_on_pcm_frame, &call_info);

  END_TO_CALL_TSFN(call_info, "Failed to call extension on_pcm_frame().");

  if (!rc || !call_info.js_result) {
    rte_on_pcm_frame_done(rte, NULL);
  }
}

static void invoke_extension_js_on_image_frame(napi_env env, napi_value fn,
                                               UNUSED void* context,
                                               void* data) {
  on_xxx_call_info_t* call_info = data;

  napi_value js_extension = NULL;
  napi_value js_rte = NULL;
  GET_JS_EXTENSION_AND_JS_RTE_FROM_BRIDGE(env, call_info->extension_bridge,
                                          js_extension, js_rte);

  napi_value js_image_frame = rte_nodejs_image_frame_wrap(env, call_info->msg);
  assert(js_image_frame);

  napi_value result = NULL;
  napi_value argv[] = {js_rte, js_image_frame};
  napi_status status =
      napi_call_function(env, js_extension, fn, 2, argv, &result);
  GOTO_LABEL_IF_NAPI_FAIL(error, status == napi_ok,
                          "Failed to call JS extension on_image_frame().");

  goto done;

error:
  call_info->js_result = false;

done:
  rte_nodejs_sem_signal(&call_info->sem);
}

static void proxy_on_image_frame(rte_extension_t* extension, UNUSED rte_t* rte,
                                 rte_sharedptr_t* frame) {
  rte_nodejs_extension_t* extension_bridge =
      rte_extension_get_me_in_target_lang(extension);
  assert(extension_bridge &&
         // RTE_NOLINTNEXTLINE(thread-check)
         // thread-check: in order to maintain thread safety, we use semaphore
         // below to prevent JS main thread and the RTE extension thread access
         // the extension bridge at the same time.
         rte_nodejs_extension_check_integrity(extension_bridge, false));

  on_xxx_call_info_t call_info = {
      extension_bridge,
      frame,
      {0},
      .js_result = true,
  };

  PREPARE_TO_CALL_TSFN(call_info);

  bool rc = rte_nodejs_threadsafe_function_invoke(
      extension_bridge->js_on_image_frame, &call_info);

  END_TO_CALL_TSFN(call_info, "Failed to call extension on_image_frame().");

  if (!rc || !call_info.js_result) {
    rte_on_image_frame_done(rte, NULL);
  }
}

static void rte_nodejs_extension_detach_callbacks(
    rte_nodejs_extension_t* extension_bridge) {
  assert(extension_bridge &&
         rte_nodejs_extension_check_integrity(extension_bridge, true));

  rte_nodejs_threadsafe_function_dec_rc(extension_bridge->js_on_init);
  rte_nodejs_threadsafe_function_dec_rc(extension_bridge->js_on_start);
  rte_nodejs_threadsafe_function_dec_rc(extension_bridge->js_on_stop);
  rte_nodejs_threadsafe_function_dec_rc(extension_bridge->js_on_cmd);
  rte_nodejs_threadsafe_function_dec_rc(extension_bridge->js_on_data);
  rte_nodejs_threadsafe_function_dec_rc(extension_bridge->js_on_pcm_frame);
  rte_nodejs_threadsafe_function_dec_rc(extension_bridge->js_on_image_frame);
}

static void rte_nodejs_extension_finalize(napi_env env, void* data,
                                          UNUSED void* hint) {
  assert(env);

  rte_nodejs_extension_t* extension_bridge = data;
  assert(extension_bridge &&
         rte_nodejs_extension_check_integrity(extension_bridge, true));

  ZF_LOGV("RTE JS extension is finalized.");

  // Unreference the JS rte instance to prevent it from not being garbage
  // collected.
  uint32_t rte_ref_count = 0;
  UNUSED napi_status status =
      napi_reference_unref(env, extension_bridge->js_rte_ref, &rte_ref_count);
  ASSERT_IF_NAPI_FAIL(status == napi_ok && rte_ref_count == 0,
                      "Failed to unref JS rte reference in extension: %d",
                      status);

  // The JS extension is finalized, so delete the reference to it.
  napi_delete_reference(env, extension_bridge->js_ref);

  // Mark the JS extension does not exist.
  extension_bridge->js_ref = NULL;

  // The JS extension is finalized, so decrease the reference count to the
  // extension bridge to indicate that one of the ownership of the extension
  // bridge will not be hold by the JS extension.
  //
  //                             addon
  //                               |
  //                               v
  //   JS extension         extension_bridge         C extension
  //         o      ==x==>         o         <==          o
  //                               ^
  //                               |
  //                             group
  //
  rte_sharedptr_dec_rc(extension_bridge->shared_ptr);
}

static void rte_nodejs_extension_attach_callbacks(
    napi_env env, rte_nodejs_extension_t* self) {
  assert(env && self && rte_nodejs_extension_check_integrity(self, true));

  napi_value js_extension = NULL;
  UNUSED napi_status status =
      napi_get_reference_value(env, self->js_ref, &js_extension);
  ASSERT_IF_NAPI_FAIL(status == napi_ok && js_extension != NULL,
                      "Failed to get JS extension: %d", status);

  napi_value js_cb_on_init =
      rte_nodejs_get_property(env, js_extension, "_onInit");
  CREATE_JS_CALLBACK(self->js_on_init, env, "Extension::onInit", js_cb_on_init,
                     invoke_extension_js_on_init);

  napi_value js_cb_on_start =
      rte_nodejs_get_property(env, js_extension, "_onStart");
  CREATE_JS_CALLBACK(self->js_on_start, env, "Extension::onStart",
                     js_cb_on_start, invoke_extension_js_on_start);

  napi_value js_cb_on_stop =
      rte_nodejs_get_property(env, js_extension, "_onStop");
  CREATE_JS_CALLBACK(self->js_on_stop, env, "Extension::onStop", js_cb_on_stop,
                     invoke_extension_js_on_stop);

  napi_value js_cb_on_cmd =
      rte_nodejs_get_property(env, js_extension, "_onCmd");
  CREATE_JS_CALLBACK(self->js_on_cmd, env, "Extension::onCmd", js_cb_on_cmd,
                     invoke_extension_js_on_cmd);

  napi_value js_cb_on_data =
      rte_nodejs_get_property(env, js_extension, "_onData");
  CREATE_JS_CALLBACK(self->js_on_data, env, "Extension::onData", js_cb_on_data,
                     invoke_extension_js_on_data);

  napi_value js_cb_on_pcm_frame =
      rte_nodejs_get_property(env, js_extension, "_onPcmFrame");
  CREATE_JS_CALLBACK(self->js_on_pcm_frame, env, "Extension::onPcmFrame",
                     js_cb_on_pcm_frame, invoke_extension_js_on_pcm_frame);

  napi_value js_cb_on_image_frame =
      rte_nodejs_get_property(env, js_extension, "_onImageFrame");
  CREATE_JS_CALLBACK(self->js_on_image_frame, env, "Extension::onImageFrame",
                     js_cb_on_image_frame, invoke_extension_js_on_image_frame);
}

static void rte_nodejs_extension_destroy(rte_nodejs_extension_t* self) {
  assert(self && rte_nodejs_extension_check_integrity(self, true));

  rte_nodejs_extension_detach_callbacks(self);

  rte_sanitizer_thread_check_deinit(&self->thread_check);
  rte_free(self);
}

static napi_value rte_nodejs_extension_create(napi_env env,
                                              napi_callback_info info) {
  assert(env && info);

  rte_string_t name;
  rte_string_init(&name);

  const size_t argc = 2;
  napi_value args[argc];  // this, name
  if (!rte_nodejs_get_args(env, info, args, argc)) {
    goto done;
  }

  if (!rte_nodejs_get_str(env, args[1], &name)) {
    goto done;
  }

  rte_nodejs_extension_t* extension_bridge =
      rte_malloc(sizeof(rte_nodejs_extension_t));
  assert(extension_bridge);

  memset(extension_bridge, 0, sizeof(rte_nodejs_extension_t));
  rte_signature_set(&extension_bridge->signature,
                    RTE_NODEJS_EXTENSION_SIGNATURE);
  rte_sanitizer_thread_check_init_with_current_thread(
      &extension_bridge->thread_check);

  napi_status status =
      napi_wrap(env, args[0], extension_bridge, rte_nodejs_extension_finalize,
                NULL, &extension_bridge->js_ref);
  GOTO_LABEL_IF_NAPI_FAIL(error, status == napi_ok,
                          "Failed to bind JS extension & bridge: %d", status);

  // There are 4 shared_ptr references to this extension_bridge:
  // - JS extension
  // - C extension
  // - addon
  // - JS extension_group
  //
  // Note: The references from 'addon' and 'group' should not exist at the same
  // time, because extension could only be created by a JS extension group _OR_
  // by an addon.
  //
  //                           addon
  //                             |
  //                             v
  //   JS extension       extension_bridge         C extension
  //         o       ==>         o          <==         o
  //                             ^
  //                             |
  //                     JS extension group
  //
  // - rte_sharedptr_create() would create the 1st reference.
  // - rte_sharedptr_inc_rc() would create the 2nd reference.
  // - when addon creates the extension, it would create the 3rd reference.
  // - when JS extension group creates the extension, it would create the 4th
  // reference.
  //
  // When one of the sources is destroyed or destroys the extension, it would
  // release 1 reference to the extension_bridge, and when all the sources are
  // done, the extension_bridge will be destroyed, too. And that is what we
  // want.
  //
  // The 1 reference count created from rte_sharedptr_create() is of the JS
  // extension.
  extension_bridge->shared_ptr =
      rte_sharedptr_create(extension_bridge, rte_nodejs_extension_destroy);

  // Increase the reference count of the JS extension, so that it will not be
  // garbage collected, and it must be paired with a napi_reference_unref() when
  // the C extension is destroyed, so that the JS extension could be garbage
  // collected successfully.
  status = napi_reference_ref(env, extension_bridge->js_ref, NULL);
  GOTO_LABEL_IF_NAPI_FAIL(error, status == napi_ok,
                          "Failed to get JS extension: %d", status);

  // Create corresponding C extension.
  extension_bridge->c_extension = rte_extension_create(
      rte_string_c_str(&name), proxy_on_init, proxy_on_start, proxy_on_stop,
      proxy_on_cmd, proxy_on_data, proxy_on_pcm_frame, proxy_on_image_frame,
      NULL);
  rte_extension_set_me_in_target_lang(extension_bridge->c_extension,
                                      extension_bridge);
  // The 1 reference count created from rte_sharedptr_inc_rc() is of the C
  // extension.
  rte_sharedptr_inc_rc(extension_bridge->shared_ptr);

  rte_nodejs_extension_attach_callbacks(env, extension_bridge);

  goto done;

error:
  if (extension_bridge) {
    rte_free(extension_bridge);
  }

done:
  rte_string_deinit(&name);

  return UNDEFINED(env);
}

static void rte_nodejs_extension_destroy_c_part(napi_env env,
                                                rte_nodejs_extension_t* self) {
  assert(self && rte_nodejs_extension_check_integrity(self, true));
  assert(self->c_extension);

  rte_nodejs_extension_release_callbacks(env, self);

  // The C extension is going to be closed, so we decrease the reference count
  // of the extension_bridge to indicate that there is no reference from the C
  // extension.
  //
  //                           addon
  //                             |
  //                             v
  //   JS extension       extension_bridge         C extension
  //         o       ==>         o         <==x==       o
  //                             ^
  //                             |
  //                           group
  //
  rte_sharedptr_dec_rc(self->shared_ptr);

  // Mark the C extension doesn't exist.
  self->c_extension = NULL;
}

/**
 * @brief This function would be called when the C extension ends.
 */
static napi_value rte_nodejs_extension_deinit(napi_env env,
                                              napi_callback_info info) {
  assert(env);

  napi_value args[1];  // this(extension)
  if (!rte_nodejs_get_args(env, info, args, 1)) {
    return UNDEFINED(env);
  }

  rte_nodejs_extension_t* extension_bridge = NULL;
  napi_status status = napi_unwrap(env, args[0], (void**)&extension_bridge);
  RETURN_UNDEFINED_IF_NAPI_FAIL(status == napi_ok && extension_bridge != NULL,
                                "Failed to get extension bridge: %d", status);

  rte_nodejs_extension_destroy_c_part(env, extension_bridge);

  return UNDEFINED(env);
}

napi_value rte_nodejs_extension_module_init(napi_env env, napi_value exports) {
  assert(env && exports);

  EXPORT_FUNC(env, exports, rte_nodejs_extension_create);
  EXPORT_FUNC(env, exports, rte_nodejs_extension_deinit);

  return exports;
}