feat(wasm): main branch

src/facade/CMakeLists.txt

@@ -1,6 +1,6 @@
 add_library(dfly_facade conn_context.cc dragonfly_listener.cc dragonfly_connection.cc facade.cc
             memcache_parser.cc redis_parser.cc reply_builder.cc op_status.cc service_interface.cc
-            reply_capture.cc resp_expr.cc cmd_arg_parser.cc tls_error.cc)
+            reply_capture.cc resp_expr.cc cmd_arg_parser.cc tls_error.cc reply_formats.cc)
 
 if (DF_USE_SSL)
   set(TLS_LIB tls_lib)

src/facade/reply_capture.h

@@ -14,14 +14,10 @@
 
 namespace facade {
 
-struct CaptureVisitor;
-
 // CapturingReplyBuilder allows capturing replies and retrieveing them with Take().
 // Those replies can be stored standalone and sent with
 // CapturingReplyBuilder::Apply() to another reply builder.
 class CapturingReplyBuilder : public RedisReplyBuilder {
-  friend struct CaptureVisitor;
-
  public:
   void SendError(std::string_view str, std::string_view type = {}) override;
   void SendError(ErrorReply error) override;

src/facade/reply_formats.cc

@@ -0,0 +1,177 @@
+#include "facade/reply_formats.h"
+
+#include <absl/strings/str_join.h>
+
+#include "absl/strings/str_cat.h"
+#include "base/logging.h"
+#include "facade/reply_capture.h"
+
+namespace facade {
+
+namespace {
+
+using namespace std;
+
+// Escape a string so that it is legal to print it in JSON text.
+std::string JsonEscape(string_view input) {
+  auto hex_digit = [](unsigned c) -> char {
+    DCHECK_LT(c, 0xFu);
+    return c < 10 ? c + '0' : c - 10 + 'a';
+  };
+
+  string out;
+  out.reserve(input.size() + 2);
+  out.push_back('\"');
+
+  auto p = input.begin();
+  auto e = input.end();
+
+  while (p < e) {
+    uint8_t c = *p;
+    if (c == '\\' || c == '\"') {
+      out.push_back('\\');
+      out.push_back(*p++);
+    } else if (c <= 0x1f) {
+      switch (c) {
+        case '\b':
+          out.append("\\b");
+          p++;
+          break;
+        case '\f':
+          out.append("\\f");
+          p++;
+          break;
+        case '\n':
+          out.append("\\n");
+          p++;
+          break;
+        case '\r':
+          out.append("\\r");
+          p++;
+          break;
+        case '\t':
+          out.append("\\t");
+          p++;
+          break;
+        default:
+          // this condition captures non readable chars with value < 32,
+          // so size = 1 byte (e.g control chars).
+          out.append("\\u00");
+          out.push_back(hex_digit((c & 0xf0) >> 4));
+          out.push_back(hex_digit(c & 0xf));
+          p++;
+      }
+    } else {
+      out.push_back(*p++);
+    }
+  }
+
+  out.push_back('\"');
+  return out;
+}
+
+struct CaptureVisitor {
+  CaptureVisitor() {
+    str = R"({"result":)";
+  }
+
+  void operator()(monostate) {
+  }
+
+  void operator()(long v) {
+    absl::StrAppend(&str, v);
+  }
+
+  void operator()(double v) {
+    absl::StrAppend(&str, v);
+  }
+
+  void operator()(const CapturingReplyBuilder::SimpleString& ss) {
+    absl::StrAppend(&str, "\"", ss, "\"");
+  }
+
+  void operator()(const CapturingReplyBuilder::BulkString& bs) {
+    absl::StrAppend(&str, JsonEscape(bs));
+  }
+
+  void operator()(CapturingReplyBuilder::Null) {
+    absl::StrAppend(&str, "null");
+  }
+
+  void operator()(CapturingReplyBuilder::Error err) {
+    str = absl::StrCat(R"({"error": ")", err.first, "\"");
+  }
+
+  void operator()(facade::OpStatus status) {
+    absl::StrAppend(&str, "\"", facade::StatusToMsg(status), "\"");
+  }
+
+  void operator()(const CapturingReplyBuilder::StrArrPayload& sa) {
+    absl::StrAppend(&str, "[");
+    for (const auto& val : sa.arr) {
+      absl::StrAppend(&str, JsonEscape(val), ",");
+    }
+    if (sa.arr.size())
+      str.pop_back();
+    absl::StrAppend(&str, "]");
+  }
+
+  void operator()(const unique_ptr<CapturingReplyBuilder::CollectionPayload>& cp) {
+    if (!cp) {
+      absl::StrAppend(&str, "null");
+      return;
+    }
+    if (cp->len == 0 && cp->type == facade::RedisReplyBuilder::ARRAY) {
+      absl::StrAppend(&str, "[]");
+      return;
+    }
+
+    absl::StrAppend(&str, "[");
+    for (auto& pl : cp->arr) {
+      visit(*this, std::move(pl));
+    }
+  }
+
+  void operator()(const facade::SinkReplyBuilder::MGetResponse& resp) {
+    absl::StrAppend(&str, "[");
+    for (const auto& val : resp.resp_arr) {
+      if (val) {
+        absl::StrAppend(&str, JsonEscape(val->value), ",");
+      } else {
+        absl::StrAppend(&str, "null,");
+      }
+    }
+
+    if (resp.resp_arr.size())
+      str.pop_back();
+    absl::StrAppend(&str, "]");
+  }
+
+  void operator()(const CapturingReplyBuilder::ScoredArray& sarr) {
+    absl::StrAppend(&str, "[");
+    for (const auto& [key, score] : sarr.arr) {
+      absl::StrAppend(&str, "{", JsonEscape(key), ":", score, "},");
+    }
+    if (sarr.arr.size() > 0) {
+      str.pop_back();
+    }
+    absl::StrAppend(&str, "]");
+  }
+
+  string Take() {
+    absl::StrAppend(&str, "}\r\n");
+    return std::move(str);
+  }
+
+  string str;
+};
+
+}  // namespace
+
+std::string FormatToJson(CapturingReplyBuilder::Payload&& payload) {
+  CaptureVisitor visitor{};
+  std::visit(visitor, payload);
+  return visitor.Take();
+}
+
+};  // namespace facade

src/facade/reply_formats.h

@@ -0,0 +1,9 @@
+#pragma once
+
+#include "facade/reply_capture.h"
+
+namespace facade {
+
+std::string FormatToJson(facade::CapturingReplyBuilder::Payload&& value);
+
+};

src/server/CMakeLists.txt

@@ -54,7 +54,8 @@ add_library(dragonfly_lib bloom_family.cc engine_shard_set.cc channel_store.cc
             cluster/cluster_family.cc cluster/incoming_slot_migration.cc
             cluster/outgoing_slot_migration.cc cluster/cluster_defs.cc
             acl/user.cc acl/user_registry.cc acl/acl_family.cc
-            acl/validator.cc acl/helpers.cc)
+            acl/validator.cc acl/helpers.cc
+            wasm/wasm_registry.cc wasm/wasm_family.cc)
 
 if (DF_ENABLE_MEMORY_TRACKING)
   target_compile_definitions(dragonfly_lib PRIVATE DFLY_ENABLE_MEMORY_TRACKING)
@@ -77,6 +78,14 @@ cxx_link(dragonfly_lib dfly_transaction dfly_facade redis_lib awsv2_lib jsonpath
          http_client_lib absl::random_random TRDP::jsoncons ${ZSTD_LIB} TRDP::lz4
          TRDP::croncpp TRDP::flatbuffers)
 
+# Better way to integrate
+# https://github.com/corrosion-rs/corrosion
+target_include_directories(dfly_transaction PUBLIC ${CMAKE_SOURCE_DIR}/c-api/include)
+
+add_library(wasmtime STATIC IMPORTED)
+target_include_directories(dragonfly_lib PRIVATE ${CMAKE_SOURCE_DIR}/c-api/include)
+target_link_libraries(dragonfly_lib ${CMAKE_SOURCE_DIR}/c-api/lib/libwasmtime.a)
+
 if (DF_USE_SSL)
   set(TLS_LIB tls_lib)
   target_compile_definitions(dragonfly_lib PRIVATE DFLY_USE_SSL)

src/server/acl/acl_commands_def.h

@@ -37,6 +37,7 @@ enum AclCat {
   SCRIPTING = 1ULL << 20,
 
   // Extensions
+  WASM = 1ULL << 27,
   BLOOM = 1ULL << 28,
   FT_SEARCH = 1ULL << 29,
   THROTTLE = 1ULL << 30,
@@ -67,6 +68,7 @@ inline const absl::flat_hash_map<std::string_view, uint32_t> CATEGORY_INDEX_TABL
     {"CONNECTION", CONNECTION},
     {"TRANSACTION", TRANSACTION},
     {"SCRIPTING", SCRIPTING},
+    {"WASM", WASM},
     {"BLOOM", BLOOM},
     {"FT_SEARCH", FT_SEARCH},
     {"THROTTLE", THROTTLE},
@@ -81,7 +83,7 @@ inline const std::vector<std::string> REVERSE_CATEGORY_INDEX_TABLE{
     "KEYSPACE",  "READ",      "WRITE",     "SET",       "SORTEDSET",  "LIST",        "HASH",
     "STRING",    "BITMAP",    "HYPERLOG",  "GEO",       "STREAM",     "PUBSUB",      "ADMIN",
     "FAST",      "SLOW",      "BLOCKING",  "DANGEROUS", "CONNECTION", "TRANSACTION", "SCRIPTING",
-    "_RESERVED", "_RESERVED", "_RESERVED", "_RESERVED", "_RESERVED",  "_RESERVED",   "_RESERVED",
+    "_RESERVED", "_RESERVED", "_RESERVED", "_RESERVED", "_RESERVED",  "_RESERVED",   "WASM",
     "BLOOM",     "FT_SEARCH", "THROTTLE",  "JSON"};
 
 using RevCommandField = std::vector<std::string>;